From 464546b0b9af29c91511d70fa2db187d74ee344a Mon Sep 17 00:00:00 2001 From: Amar3tto Date: Wed, 24 Jun 2026 20:25:49 +0000 Subject: [PATCH 1/3] Set Dataflow container to release version. --- runners/google-cloud-dataflow-java/build.gradle | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/runners/google-cloud-dataflow-java/build.gradle b/runners/google-cloud-dataflow-java/build.gradle index 015825fd6de1..7e9b3d03da97 100644 --- a/runners/google-cloud-dataflow-java/build.gradle +++ b/runners/google-cloud-dataflow-java/build.gradle @@ -52,8 +52,8 @@ evaluationDependsOn(":sdks:java:container:java11") ext.dataflowLegacyEnvironmentMajorVersion = '8' ext.dataflowFnapiEnvironmentMajorVersion = '8' -ext.dataflowLegacyContainerVersion = 'beam-master-20260601' -ext.dataflowFnapiContainerVersion = 'beam-master-20260601' +ext.dataflowLegacyContainerVersion = '2.75.0' +ext.dataflowFnapiContainerVersion = '2.75.0' ext.dataflowContainerBaseRepository = 'gcr.io/cloud-dataflow/v1beta3' processResources { From 1ec0a6b3f09257bd753977b34ab2f479ea9c986a Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 24 Jun 2026 15:55:31 -0400 Subject: [PATCH 2/3] Fix race condition in DirectRunner executor shutdown Previously, in ExecutorServiceParallelExecutor, if an exception occurred during registry cleanup (such as a timeout inside DoFn teardown), the pipeline state was transitioned to terminal before the exception was queued in `visibleUpdates`. This introduced a race condition where `waitUntilFinish()` could detect the terminal state and exit successfully before the exception was offered to the updates queue, swallowing the exception. This caused tests like `CallTest.givenTeardownTimeout_throwsError` to fail since they expected the pipeline to throw an exception. This change swaps the order so that the exception is posted to `visibleUpdates` before updating the pipeline state to terminal, ensuring the exception is always propagated. --- .../runners/direct/ExecutorServiceParallelExecutor.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java index 95cadef7afdb..43c4356ee3b2 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java @@ -348,9 +348,9 @@ private void shutdownIfNecessary(State newState) { } catch (final Exception e) { errors.add(e); } - pipelineState.compareAndSet(State.RUNNING, newState); // ensure we hit a terminal node + IllegalStateException exception = null; if (!errors.isEmpty()) { - final IllegalStateException exception = + exception = new IllegalStateException( "Error" + (errors.size() == 1 ? "" : "s") @@ -359,6 +359,9 @@ private void shutdownIfNecessary(State newState) { .map(Exception::getMessage) .collect(Collectors.joining("\n- ", "- ", ""))); visibleUpdates.failed(exception); + } + pipelineState.compareAndSet(State.RUNNING, newState); // ensure we hit a terminal node + if (exception != null) { throw exception; } } From 32134347a8a0fdd7142d78189e4716dd548cd628 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 24 Jun 2026 17:34:17 -0400 Subject: [PATCH 3/3] Apply suggested fix --- .../ExecutorServiceParallelExecutor.java | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java index 43c4356ee3b2..37cff06a267f 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java @@ -349,18 +349,21 @@ private void shutdownIfNecessary(State newState) { errors.add(e); } IllegalStateException exception = null; - if (!errors.isEmpty()) { - exception = - new IllegalStateException( - "Error" - + (errors.size() == 1 ? "" : "s") - + " during executor shutdown:\n" - + errors.stream() - .map(Exception::getMessage) - .collect(Collectors.joining("\n- ", "- ", ""))); - visibleUpdates.failed(exception); + try { + if (!errors.isEmpty()) { + exception = + new IllegalStateException( + "Error" + + (errors.size() == 1 ? "" : "s") + + " occurred during pipeline execution:\\n" + + errors.stream() + .map(e -> e.getMessage() == null ? e.getClass().getName() : e.getMessage()) + .collect(Collectors.joining("\\n- ", "- ", ""))); + visibleUpdates.failed(exception); + } + } finally { + pipelineState.compareAndSet(State.RUNNING, newState); // ensure we hit a terminal node } - pipelineState.compareAndSet(State.RUNNING, newState); // ensure we hit a terminal node if (exception != null) { throw exception; }