Update BQIO to a single scheduled executor service reduce threads #23234

johnjcasey · 2022-09-14T18:12:07Z

fixes #21368

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

…ogation

github-actions · 2022-09-15T03:14:31Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

johnjcasey · 2022-09-15T13:48:18Z

run java precommit

github-actions · 2022-09-15T20:01:05Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @kileys for label java.
R: @pabloem for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

github-actions · 2022-09-24T12:13:59Z

Reminder, please take a look at this pr: @kileys @pabloem

github-actions · 2022-09-28T12:14:04Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @robertwb for label java.
R: @chamikaramj for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

johnjcasey · 2022-09-29T13:36:07Z

Run Java_Examples_Dataflow_Java17 PreCommit

johnjcasey · 2022-09-29T15:59:25Z

r: @lukecwik can you take a look at this? I believe we will need to have a separate executor service to preserve backwards compatibility, because if someone provides a non-scheduled executor service then BQIO will fail

lukecwik · 2022-09-29T16:52:04Z

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

+       */
+
+      return Executors.newScheduledThreadPool(
+          Math.max(4, Runtime.getRuntime().availableProcessors()), threadFactoryBuilder.build());


What was your reasoning around choosing this as the minimum?

I dug into the default configuration in BQ. When we set the minimum to 0, threads weren't firing, so I looked to the known working configuration in BQ, and used that

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

lukecwik · 2022-09-29T17:02:07Z

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

+   */
+  @JsonIgnore
+  @Description(
+      "The ScheduledExecutorService instance to use to create threads, can be overridden to specify "


It makes a lot of sense to not declare this pipeline option in GcsOptions since the original choice of having this in GcsOptions was done a long time ago since GCS needed it but it is used in many places.

Since it is used both during pipeline construction and pipeline execution it makes sense to move it to its own options class in org/apache/beam/sdk/options/ExecutorOptions.java in sdks/java/core

makes sense, I'll refactor this

...e-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java

github-actions · 2022-09-29T17:44:19Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExecutorOptions.java

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

…ervices

lukecwik

All nits in the current review.

Since we are returning a ScheduledExecutorService now for GcsOptions#getExecutorService, do we not get test failures anymore (e.g. RemoteExecutionTest)?

In a follow-up can you migrate all usages of GcsOptions#getExecutorService to ExecutorOptions#getScheduledExecutorService?

sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExecutorOptions.java

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

lukecwik · 2022-09-29T21:57:35Z

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

  ExecutorService getExecutorService();

+  /**
+   * @deprecated use {@link ExecutorOptions#setScheduledExecutorService(ScheduledExecutorService)}


Suggested change

* @deprecated use {@link ExecutorOptions#setScheduledExecutorService(ScheduledExecutorService)}

* @deprecated use {@link ExecutorOptions#setScheduledExecutorService}

johnjcasey · 2022-09-30T17:00:28Z

All nits in the current review.

Since we are returning a ScheduledExecutorService now for GcsOptions#getExecutorService, do we not get test failures anymore (e.g. RemoteExecutionTest)?

In a follow-up can you migrate all usages of GcsOptions#getExecutorService to ExecutorOptions#getScheduledExecutorService?

I don't think we can do this, because if someone is using GcsOptions#setExecutorService, this change could break them

lukecwik · 2022-10-06T18:39:40Z

It looks like setting core pool size to int.max and allowing core thread timeout is working it is recommended to not do that, but the concern of running out of threads doesn't appear to be valid.

This does not do what you want since a corePoolSize == Integer.MAX will mean that we will want to start a new thread before checking to see if there is already an idle thread capable of doing the work. This leads to every task running on a new thread and then expiring after the timeout (which is very expensive performance wise). This is why the previous thread pool had a core pool size of 0. This forced the thread pool to first offer it to the queue and then start a new worker if all the workers are busy.

johnjcasey · 2022-10-06T20:08:07Z

ah, thats unfortunate, but you're right.

I'm not confident in implementing my own scheduler. Looking through some of the guarantees given by the interface definitions, it may be quite challenging.

johnjcasey · 2022-10-06T20:13:15Z

I don't think I can accurately implement this in < several weeks

johnjcasey · 2022-10-06T20:24:01Z

A number of the guarantees around periodic and delayed execution depend on methods that are package private, so I can't just copy the scheduled executor code either

lukecwik · 2022-10-07T17:51:45Z

A number of the guarantees around periodic and delayed execution depend on methods that are package private, so I can't just copy the scheduled executor code either

Give me a couple of hours to take a deeper look.

lukecwik · 2022-10-07T23:46:07Z

I think I got an implementation for an unbounded ScheduledExecutorService with the properties that we want.
PTAL at #23545

lukecwik · 2022-10-13T17:04:41Z

@johnjcasey Ping me when you have merged your changes with the new UnboundedScheduledExecutorService.

…educe-gax-threads # Conflicts: # sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

lukecwik

All my comments are minor, please fix and merge when addressed.

lukecwik · 2022-10-14T19:48:01Z

runners/google-cloud-dataflow-java/build.gradle

@@ -287,7 +287,8 @@ def buildAndPushDockerJavaContainer = tasks.register("buildAndPushDockerJavaCont
      commandLine "docker", "tag", "${defaultDockerImageName}", "${dockerJavaImageName}"
    }
    exec {
-      commandLine "gcloud", "docker", "--", "push", "${dockerJavaImageName}"
+//      commandLine "gcloud", "docker", "--", "push", "${dockerJavaImageName}"


is this necessary?

this was commited in error

lukecwik · 2022-10-14T19:49:22Z

sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExecutorOptions.java

-import java.util.concurrent.TimeUnit;
-import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.MoreExecutors;
-import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.beam.sdk.util.UnboundedScheduledExecutorService;

 public interface ExecutorOptions extends PipelineOptions {



comment and @Description needs to be updated since the core pool size is irrelevant now.

lukecwik · 2022-10-14T19:50:18Z

sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExecutorOptions.java

+import java.util.concurrent.ScheduledExecutorService;
+import org.apache.beam.sdk.util.UnboundedScheduledExecutorService;
+
+public interface ExecutorOptions extends PipelineOptions {


class comment?

lukecwik · 2022-10-14T19:52:02Z

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

-      /* The SDK requires an unbounded thread pool because a step may create X writers
-       * each requiring their own thread to perform the writes otherwise a writer may
-       * block causing deadlock for the step because the writers buffer is full.
-       * Also, the MapTaskExecutor launches the steps in reverse order and completes
-       * them in forward order thus requiring enough threads so that each step's writers
-       * can be active.
-       */


Move this comment into ScheduledExecutorServiceFactory just above new UnboundedScheduledExecutorService()

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

…org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

reuvenlax · 2022-10-22T02:36:59Z

This PR causes the BigQuery client to hang on shutdown.

…eads (apache#23234)" This reverts commit 8e2431c.

…eads (#23234)" (#23793) This reverts commit 8e2431c.

…duce threads (apache#23234)" (apache#23793)" This reverts commit 01da3fc.

…eads (apache#23234)" (apache#23793) This reverts commit 8e2431c. (cherry picked from commit 01da3fc)

* Merge pull request #23795: Revert 23234: issue #23794 * Revert "Update BQIO to a single scheduled executor service reduce threads (#23234)" (#23793) This reverts commit 8e2431c. (cherry picked from commit 01da3fc) * Merge pull request #23795: Revert 23234: issue #23794 (cherry picked from commit d38f577) Co-authored-by: reuvenlax <relax@google.com>

mdlnr · 2022-11-07T10:42:00Z

This PR causes the BigQuery client to hang on shutdown.

@reuvenlax We observed the behaviour of hanging threads in our Dataflow jobs during StreamWriter.close with Beam Version 2.41.0. So maybe the PR does not cause this problem but might make it more visible.

In our pipelines we see continously increasing number of threads with stack traces like these:

"pool-3-thread-61" #1767 prio=5 os_prio=0 cpu=0.22ms elapsed=337556.83s tid=0x00007f2d982459d0 nid=0x6f7 in Object.wait()  [0x00007f2d886f1000]
   java.lang.Thread.State: WAITING (on object monitor)
	at java.lang.Object.wait(java.base@17.0.2/Native Method)
	- waiting on <no object reference available>
	at java.lang.Thread.join(java.base@17.0.2/Thread.java:1304)
	- locked <merged>(a java.lang.Thread)
	at java.lang.Thread.join(java.base@17.0.2/Thread.java:1372)
	at com.google.cloud.bigquery.storage.v1.StreamWriter.close(StreamWriter.java:369)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$DatasetServiceImpl$1.close(BigQueryServicesImpl.java:1339)
	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWritesShardedRecords$$Lambda$692/0x00000008015dd4c8.run(Unknown Source)
	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWritesShardedRecords.lambda$runAsyncIgnoreFailure$1(StorageApiWritesShardedRecords.java:138)
	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWritesShardedRecords$$Lambda$685/0x00000008015d6b58.run(Unknown Source)
	at java.util.concurrent.Executors$RunnableAdapter.call(java.base@17.0.2/Executors.java:539)
	at java.util.concurrent.FutureTask.run(java.base@17.0.2/FutureTask.java:264)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.2/ThreadPoolExecutor.java:1136)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.2/ThreadPoolExecutor.java:635)
	at java.lang.Thread.run(java.base@17.0.2/Thread.java:833)

We think this is related to using a pretty old version of the dependency com.google.cloud:google-cloud-bigquerystorage:2.12.2. The library has since made many changes in the StreamWriter class which could fix this issue. Is there anything that prevents updating a newer released version?

reuvenlax · 2022-11-07T18:49:14Z

Shutdown hangs were definitely being caused by that PR. Note that the hang seen in that PR was shutting down the RPC client, not shutting down the StreamWriter. This appears to be a different issue. FYI the current head Beam is depending on 2.24.2.

…

On Mon, Nov 7, 2022 at 2:42 AM Thorsten Madlener ***@***.***> wrote: This PR causes the BigQuery client to hang on shutdown. @reuvenlax <https://github.com/reuvenlax> We observed the behaviour of hanging threads in our Dataflow jobs during StreamWriter.close with Beam Version 1.41.0. So maybe the PR does not cause this problem but might make it more visible. In our pipelines we see continously increasing number of threads with stack traces like these: "pool-3-thread-61" #1767 prio=5 os_prio=0 cpu=0.22ms elapsed=337556.83s tid=0x00007f2d982459d0 nid=0x6f7 in Object.wait() [0x00007f2d886f1000] java.lang.Thread.State: WAITING (on object monitor) at ***@***.***/Native Method) - waiting on <no object reference available> at ***@***.***/Thread.java:1304) - locked <merged>(a java.lang.Thread) at ***@***.***/Thread.java:1372) at com.google.cloud.bigquery.storage.v1.StreamWriter.close(StreamWriter.java:369) at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$DatasetServiceImpl$1.close(BigQueryServicesImpl.java:1339) at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWritesShardedRecords$$Lambda$692/0x00000008015dd4c8.run(Unknown Source) at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWritesShardedRecords.lambda$runAsyncIgnoreFailure$1(StorageApiWritesShardedRecords.java:138) at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWritesShardedRecords$$Lambda$685/0x00000008015d6b58.run(Unknown Source) at ***@***.***/Executors.java:539) at ***@***.***/FutureTask.java:264) at ***@***.***/ThreadPoolExecutor.java:1136) at ***@***.***/ThreadPoolExecutor.java:635) at ***@***.***/Thread.java:833) We think this is related to using a pretty old version of the dependency com.google.cloud:google-cloud-bigquerystorage:2.12.2. The library has since made many changes in the StreamWriter class which could fix this issue. Is there anything that prevents updating a newer released version? — Reply to this email directly, view it on GitHub <#23234 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVNYIU64P4TB2UNE7XTWHDMIHANCNFSM6AAAAAAQMVGVQM> . You are receiving this because you were mentioned.Message ID: ***@***.***>

Update to a single scheduled executor service to mitigate thread prop…

a7fb056

…ogation

johnjcasey changed the title ~~Update to a single scheduled executor service reduce threads~~ Update BQIO to a single scheduled executor service reduce threads Sep 14, 2022

run spotless

1e062e2

github-actions bot added extensions gcp io java labels Sep 14, 2022

run spotless

f2c95a9

Provide separate scheduled executor

5b81a10

github-actions bot added the Next Action: Reviewers label Sep 15, 2022

try scheduled thread pool again

97fe2c2

github-actions bot added the slow-review label Sep 24, 2022

github-actions bot removed the slow-review label Sep 28, 2022

johnjcasey added 3 commits September 28, 2022 14:38

Attempt to mimic Bigquery default scheduled thread pool

e606856

run spotless

62b4d5d

update docs

1de7feb

lukecwik requested changes Sep 29, 2022

View reviewed changes

Refactor Scheduled executor service to its own options file

45fa683

lukecwik requested changes Sep 29, 2022

View reviewed changes

clean up comments, don't accidentally create two scheduled executor s…

d68b2a1

…ervices

lukecwik approved these changes Sep 29, 2022

View reviewed changes

add . for checkstyle

3446728

github-actions bot added dataflow runners labels Oct 6, 2022

lukecwik mentioned this pull request Oct 13, 2022

Io jms fix ack message checkpoint #22932

Merged

1 task

johnjcasey added 2 commits October 14, 2022 14:30

Merge remote-tracking branch 'origin/master' into bugfix/Beam-21368-r…

e8b1301

…educe-gax-threads # Conflicts: # sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

Update ExecutorOptions to use @lukecwik's unbounded scheduled executor

8d9bb75

lukecwik approved these changes Oct 14, 2022

View reviewed changes

johnjcasey added 2 commits October 14, 2022 15:59

revert change to runners/google-cloud-dataflow-java/build.gradle

a59e30a

Update comments on options files

eb88210

github-actions bot removed dataflow runners labels Oct 14, 2022

lukecwik reviewed Oct 14, 2022

View reviewed changes

...cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java Outdated Show resolved Hide resolved

Update sdks/java/extensions/google-cloud-platform-core/src/main/java/…

cc869f7

…org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

lukecwik merged commit 8e2431c into apache:master Oct 15, 2022

reuvenlax added a commit to reuvenlax/incubator-beam that referenced this pull request Oct 22, 2022

Revert "Update BQIO to a single scheduled executor service reduce thr…

e88c7d0

…eads (apache#23234)" This reverts commit 8e2431c.

reuvenlax added a commit that referenced this pull request Oct 22, 2022

Revert "Update BQIO to a single scheduled executor service reduce thr…

01da3fc

…eads (#23234)" (#23793) This reverts commit 8e2431c.

reuvenlax added a commit to reuvenlax/incubator-beam that referenced this pull request Oct 22, 2022

Revert "Revert "Update BQIO to a single scheduled executor service re…

c055f83

…duce threads (apache#23234)" (apache#23793)" This reverts commit 01da3fc.

lukecwik mentioned this pull request Oct 24, 2022

Reduce number of Gax related threads, likely by providing common executor to GAX clients #21368

Closed

johnjcasey pushed a commit to johnjcasey/beam that referenced this pull request Oct 26, 2022

Revert "Update BQIO to a single scheduled executor service reduce thr…

7a75d1f

…eads (apache#23234)" (apache#23793) This reverts commit 8e2431c. (cherry picked from commit 01da3fc)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Update BQIO to a single scheduled executor service reduce threads #23234

Update BQIO to a single scheduled executor service reduce threads #23234

johnjcasey commented Sep 14, 2022

github-actions bot commented Sep 15, 2022

johnjcasey commented Sep 15, 2022

github-actions bot commented Sep 15, 2022

github-actions bot commented Sep 24, 2022

github-actions bot commented Sep 28, 2022

johnjcasey commented Sep 29, 2022

johnjcasey commented Sep 29, 2022

lukecwik Sep 29, 2022

johnjcasey Sep 29, 2022

lukecwik Sep 29, 2022

johnjcasey Sep 29, 2022

github-actions bot commented Sep 29, 2022

lukecwik left a comment

lukecwik Sep 29, 2022

johnjcasey commented Sep 30, 2022

lukecwik commented Oct 6, 2022

johnjcasey commented Oct 6, 2022

johnjcasey commented Oct 6, 2022

johnjcasey commented Oct 6, 2022

lukecwik commented Oct 7, 2022

lukecwik commented Oct 7, 2022

lukecwik commented Oct 13, 2022

lukecwik left a comment

lukecwik Oct 14, 2022

johnjcasey Oct 14, 2022

lukecwik Oct 14, 2022

lukecwik Oct 14, 2022

lukecwik Oct 14, 2022

reuvenlax commented Oct 22, 2022

mdlnr commented Nov 7, 2022 •

edited

Loading

reuvenlax commented Nov 7, 2022 via email

	* @deprecated use {@link ExecutorOptions#setScheduledExecutorService(ScheduledExecutorService)}
	* @deprecated use {@link ExecutorOptions#setScheduledExecutorService}

Update BQIO to a single scheduled executor service reduce threads #23234

Update BQIO to a single scheduled executor service reduce threads #23234

Conversation

johnjcasey commented Sep 14, 2022

fixes #21368

GitHub Actions Tests Status (on master branch)

github-actions bot commented Sep 15, 2022

johnjcasey commented Sep 15, 2022

github-actions bot commented Sep 15, 2022

github-actions bot commented Sep 24, 2022

github-actions bot commented Sep 28, 2022

johnjcasey commented Sep 29, 2022

johnjcasey commented Sep 29, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Sep 29, 2022

lukecwik left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johnjcasey commented Sep 30, 2022

lukecwik commented Oct 6, 2022

johnjcasey commented Oct 6, 2022

johnjcasey commented Oct 6, 2022

johnjcasey commented Oct 6, 2022

lukecwik commented Oct 7, 2022

lukecwik commented Oct 7, 2022

lukecwik commented Oct 13, 2022

lukecwik left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

reuvenlax commented Oct 22, 2022

mdlnr commented Nov 7, 2022 • edited Loading

reuvenlax commented Nov 7, 2022 via email

mdlnr commented Nov 7, 2022 •

edited

Loading