Do not create new Executor everytime createRunner #32272

Abacn · 2024-08-21T18:52:06Z

Please add a meaningful description for your change here

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

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 or the workflows README to see a list of phrases to trigger workflows.

Abacn · 2024-08-21T18:53:10Z

...worker/src/main/java/org/apache/beam/runners/dataflow/worker/SplittableProcessFnFactory.java

@@ -131,6 +136,11 @@ public DoFnRunner<KeyedWorkItem<byte[], KV<InputT, RestrictionT>>, OutputT> crea
        OutputManager outputManager,
        DoFnSchemaInformation doFnSchemaInformation,
        Map<String, PCollectionView<?>> sideInputMapping) {
+      if (this.ses == null) {


added debug log in both branch and run a sample job, saw

INFO 2024-08-21T18:47:53.569Z creating new sdf executor INFO 2024-08-21T18:47:55.698Z Reuse old sdf executor INFO 2024-08-21T18:47:56.812Z Reuse old sdf executor INFO 2024-08-21T18:47:57.919Z Reuse old sdf executor INFO 2024-08-21T18:47:59.030Z Reuse old sdf executor INFO 2024-08-21T18:48:00.138Z Reuse old sdf executor INFO 2024-08-21T18:48:00.699Z Reuse old sdf executor INFO 2024-08-21T18:48:01.871Z Reuse old sdf executor INFO 2024-08-21T18:48:03.044Z Reuse old sdf executor INFO 2024-08-21T18:48:04.121Z Reuse old sdf executor INFO 2024-08-21T18:48:04.715Z Reuse old sdf executor INFO 2024-08-21T18:48:05.821Z Reuse old sdf executor ...

indicating this change is effective

Abacn · 2024-08-21T19:06:25Z

internal tracker: 361080602

example dataflow job

before: 2024-08-21_11_14_45-8135795605757551727

memory usage increasing overtime (will ultimately OOM due to thread leak)

after: 2024-08-21_11_39_15-12114433774010769671

memory usage is flat

Abacn · 2024-08-21T19:09:23Z

R: @scwhittle

github-actions · 2024-08-21T19:10:37Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

Abacn · 2024-08-21T19:14:52Z

assign set of reviewers

github-actions · 2024-08-21T19:16:08Z

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

R: @chamikaramj added as fallback since no labels match configuration

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).

scwhittle · 2024-08-21T19:16:14Z

...worker/src/main/java/org/apache/beam/runners/dataflow/worker/SplittableProcessFnFactory.java

@@ -131,6 +136,11 @@ public DoFnRunner<KeyedWorkItem<byte[], KV<InputT, RestrictionT>>, OutputT> crea
        OutputManager outputManager,
        DoFnSchemaInformation doFnSchemaInformation,
        Map<String, PCollectionView<?>> sideInputMapping) {
+      if (this.ses == null) {
+        this.ses =
+            Executors.newSingleThreadScheduledExecutor(


I don't think we want a single threaded one here, because I believe the factory vends many different dofnrunner which will want some parallel threads for splits.

changed to use Executors.newScheduledThreadPool

scwhittle · 2024-08-21T19:17:25Z

...worker/src/main/java/org/apache/beam/runners/dataflow/worker/SplittableProcessFnFactory.java

@@ -131,6 +136,11 @@ public DoFnRunner<KeyedWorkItem<byte[], KV<InputT, RestrictionT>>, OutputT> crea
        OutputManager outputManager,
        DoFnSchemaInformation doFnSchemaInformation,
        Map<String, PCollectionView<?>> sideInputMapping) {
+      if (this.ses == null) {


if this factory possibly called concurrently? might need some synchronization if so?

I am not sure but now guard with AtomicReference

scwhittle · 2024-08-21T19:18:22Z

...c/main/java/org/apache/beam/runners/samza/runtime/SplittableParDoProcessKeyedElementsOp.java

@@ -137,6 +141,12 @@ public void open(
            isBounded,
            pipelineOptions);

+    if (this.ses == null) {
+      this.ses =
+          Executors.newSingleThreadScheduledExecutor(


not yet using this below

fixed the mistake

...ava/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java

...c/main/java/org/apache/beam/runners/samza/runtime/SplittableParDoProcessKeyedElementsOp.java

scwhittle

Thanks for fixing these!

scwhittle · 2024-08-21T20:39:36Z

...worker/src/main/java/org/apache/beam/runners/dataflow/worker/SplittableProcessFnFactory.java

@@ -131,6 +137,13 @@ public DoFnRunner<KeyedWorkItem<byte[], KV<InputT, RestrictionT>>, OutputT> crea
        OutputManager outputManager,
        DoFnSchemaInformation doFnSchemaInformation,
        Map<String, PCollectionView<?>> sideInputMapping) {
+      if (this.ses.get() == null) {


remembered a cleaner way to do this:

private final Supplier sesSupplier =
Suppliers.memoize(
() -> Executors.newScheduledThreadPool(
Runtime.getRuntime().availableProcessors(),
new ThreadFactoryBuilder().setNameFormat("df-sdf-executor-%d").build())));

and then just ses.get() below

thanks, good to learn, for now merging it before release cut

Do not create new Executor everytime createRunner

f6ad68f

github-actions bot added runners dataflow flink samza labels Aug 21, 2024

Abacn commented Aug 21, 2024

View reviewed changes

reset executorService after shutdown

443e38e

github-actions bot added the Next Action: Reviewers label Aug 21, 2024

scwhittle requested changes Aug 21, 2024

View reviewed changes

Abacn force-pushed the reuseexec branch from ed9338c to 3660a9d Compare August 21, 2024 19:52

Switch to use newScheduledThreadPool; guard ses with AtomicReference

d55a68e

Abacn force-pushed the reuseexec branch from 3660a9d to d55a68e Compare August 21, 2024 19:59

scwhittle requested changes Aug 21, 2024

View reviewed changes

...ava/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java Outdated Show resolved Hide resolved

...c/main/java/org/apache/beam/runners/samza/runtime/SplittableParDoProcessKeyedElementsOp.java Outdated Show resolved Hide resolved

Partially revert changes on flink and samza runner

3b08a1f

scwhittle approved these changes Aug 21, 2024

View reviewed changes

Abacn merged commit ed4c03e into apache:master Aug 21, 2024
24 checks passed

Abacn deleted the reuseexec branch August 21, 2024 21:08

lostluck added this to the 2.59.0 Release milestone Aug 27, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Do not create new Executor everytime createRunner #32272

Do not create new Executor everytime createRunner #32272

Abacn commented Aug 21, 2024

Abacn Aug 21, 2024

Abacn commented Aug 21, 2024 •

edited

Loading

Abacn commented Aug 21, 2024

github-actions bot commented Aug 21, 2024

Abacn commented Aug 21, 2024

github-actions bot commented Aug 21, 2024

scwhittle Aug 21, 2024

Abacn Aug 21, 2024

scwhittle Aug 21, 2024

Abacn Aug 21, 2024

scwhittle Aug 21, 2024

Abacn Aug 21, 2024

scwhittle left a comment

scwhittle Aug 21, 2024

Abacn Aug 21, 2024

Do not create new Executor everytime createRunner #32272

Do not create new Executor everytime createRunner #32272

Conversation

Abacn commented Aug 21, 2024

GitHub Actions Tests Status (on master branch)

Choose a reason for hiding this comment

Abacn commented Aug 21, 2024 • edited Loading

Abacn commented Aug 21, 2024

github-actions bot commented Aug 21, 2024

Abacn commented Aug 21, 2024

github-actions bot commented Aug 21, 2024

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

scwhittle left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Abacn commented Aug 21, 2024 •

edited

Loading