[GSoC 2026] Kafka Streams runner: enable ParDoTest in the ValidatesRunner suite (#39451)

Author: junaiddshaukat

runners/kafka-streams/build.gradle

@@ -92,6 +92,15 @@ def sickbayTests = [
   'org.apache.beam.sdk.transforms.GroupByKeyTest$WindowTests',
   'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testTimestampCombinerLatest',
   'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testTimestampCombinerEarliest',
+  // A DoFn whose @StartBundle throws never gets to report its error: SdkHarnessClient.newBundle
+  // sends the ProcessBundleRequest and then blocks in GrpcDataService.createOutboundAggregator
+  // waiting for the SDK harness to open its data stream, which a bundle that failed during setup
+  // never does — so the run blocks for the data service's three-minute timeout instead of
+  // surfacing the user's exception. This is shared java-fn-execution behaviour rather than
+  // anything specific to this runner; the Flink runner sickbays all of LifecycleTests and the
+  // Prism runner sickbays each of its three error tests. The @ProcessElement and @FinishBundle
+  // variants do pass here, because by then the data stream is established.
+  'org.apache.beam.sdk.transforms.ParDoTest$LifecycleTests.testParDoWithErrorInStartBatch',
 ]
 
 tasks.register("validatesRunner", Test) {
@@ -143,6 +152,7 @@ tasks.register("validatesRunner", Test) {
     includeTestsMatching 'org.apache.beam.sdk.transforms.CreateTest'
     includeTestsMatching 'org.apache.beam.sdk.transforms.FlattenTest'
     includeTestsMatching 'org.apache.beam.sdk.transforms.GroupByKeyTest*'
+    includeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest*'
     for (String test : sickbayTests) {
       excludeTestsMatching test
     }

runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunner.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.util.UUID;
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.PipelineRunner;
 import org.apache.beam.sdk.metrics.MetricResults;
@@ -47,6 +48,10 @@
  * PAssert} that silently never runs is caught by {@code TestPipeline.verifyPAssertsSucceeded}
  * comparing the success-counter metric against the number of assertions in the pipeline.
  *
+ * <p>Any other failed run is restated as a {@link PipelineExecutionException} carrying the
+ * exception the user's code threw, which is the failure Beam's contract for {@link #run} — and the
+ * tests that assert on a DoFn throwing — expect to see.
+ *
  * <p>Select it with {@code --runner=org.apache.beam.runners.kafka.streams.TestKafkaStreamsRunner}
  * in {@code beamTestPipelineOptions}.
  */
@@ -84,11 +89,31 @@ public PipelineResult run(Pipeline pipeline) {
           throw (AssertionError) current;
         }
       }
-      throw t;
+      throw pipelineExecutionExceptionFor(t);
     }
     return new TestKafkaStreamsPipelineResult(metrics);
   }
 
+  /**
+   * Restates a failed run as the {@link PipelineExecutionException} Beam's contract for {@link
+   * #run} expects, carrying the exception the user's code threw.
+   *
+   * <p>What surfaces from the run is the outermost layer of runner plumbing — a Kafka Streams
+   * {@code StreamsException} naming the processor node that failed — wrapping several layers of
+   * bundle- and Fn-API-level wrappers, with the user's exception at the bottom. Callers assert on
+   * the user's failure, so the root cause is what {@link PipelineExecutionException} is given:
+   * because {@code RuntimeException(Throwable)} derives its message from the cause, the user's
+   * message ends up on the exception that {@link #run} throws. Other runners restate failures the
+   * same way (e.g. {@code SparkPipelineResult}, {@code DirectRunner}).
+   */
+  private static PipelineExecutionException pipelineExecutionExceptionFor(Throwable t) {
+    Throwable rootCause = t;
+    while (rootCause.getCause() != null && rootCause.getCause() != rootCause) {
+      rootCause = rootCause.getCause();
+    }
+    return new PipelineExecutionException(rootCause);
+  }
+
   /** Terminal result of a test run: state {@code DONE} with the harness-reported metrics. */
   private static final class TestKafkaStreamsPipelineResult implements PipelineResult {
     private final MetricResults metrics;