Merge pull request #39115 from shunping/cp-39096

[release-2.75] Cherry-pick #39096 to the branch

Author: Amar3tto

runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java

@@ -348,17 +348,23 @@ private void shutdownIfNecessary(State newState) {
     } catch (final Exception e) {
       errors.add(e);
     }
-    pipelineState.compareAndSet(State.RUNNING, newState); // ensure we hit a terminal node
-    if (!errors.isEmpty()) {
-      final IllegalStateException exception =
-          new IllegalStateException(
-              "Error"
-                  + (errors.size() == 1 ? "" : "s")
-                  + " during executor shutdown:\n"
-                  + errors.stream()
-                      .map(Exception::getMessage)
-                      .collect(Collectors.joining("\n- ", "- ", "")));
-      visibleUpdates.failed(exception);
+    IllegalStateException exception = null;
+    try {
+      if (!errors.isEmpty()) {
+        exception =
+            new IllegalStateException(
+                "Error"
+                    + (errors.size() == 1 ? "" : "s")
+                    + " occurred during executor shutdown:\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
+    }
+    if (exception != null) {
       throw exception;
     }
   }