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.

Author: shunping

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;
     }
   }