From 0d4efb162e07dd19afb49cf7e5335b850d492f64 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Wed, 24 Jun 2026 15:55:31 -0400
Subject: [PATCH 1/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 ec83eb2852f880f937758d67dd276924150b3b4b Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Wed, 24 Jun 2026 17:34:17 -0400
Subject: [PATCH 2/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;
     }

From d76047aa7f8b79a0a00940199060696654d42db3 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Thu, 25 Jun 2026 15:26:29 -0400
Subject: [PATCH 3/3] Fix typo

---
 .../beam/runners/direct/ExecutorServiceParallelExecutor.java  | 4 ++--
 1 file changed, 2 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 37cff06a267f..603832843bb6 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
@@ -355,10 +355,10 @@ private void shutdownIfNecessary(State newState) {
             new IllegalStateException(
                 "Error"
                     + (errors.size() == 1 ? "" : "s")
-                    + " occurred during pipeline execution:\\n"
+                    + " occurred during executor shutdown:\n"
                     + errors.stream()
                         .map(e -> e.getMessage() == null ? e.getClass().getName() : e.getMessage())
-                        .collect(Collectors.joining("\\n- ", "- ", "")));
+                        .collect(Collectors.joining("\n- ", "- ", "")));
         visibleUpdates.failed(exception);
       }
     } finally {