From 2858c06de2e9f102b29e576b42509a1f4df5b9d1 Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Thu, 23 Jul 2026 12:23:10 +0500 Subject: [PATCH] [GSoC 2026] Kafka Streams runner: enable ParDoTest in the ValidatesRunner suite Adds Beam's ParDoTest suites to the validatesRunner task, taking it from 26 to 44 tests: BasicTests (6), MultipleInputsAndOutputTests (5), TimestampTests (4) and LifecycleTests (3). MultipleInputsAndOutputTests passes on the multi-output executable stages added in #39410. The remaining ParDoTest nested classes fall out on category excludes the task already declares. TestKafkaStreamsRunner.run now restates a failed run as a PipelineExecutionException carrying the exception the user's code threw, instead of letting the Kafka Streams wrapper naming the failed processor node surface. RuntimeException(Throwable) derives its message from the cause, so the user's message ends up on the thrown exception, which is what Beam's contract for run() and the tests asserting on a DoFn throwing expect to see. SparkPipelineResult and DirectRunner restate failures the same way. The existing unwrap that rethrows a failed PAssert's AssertionError still runs first. Sickbays testParDoWithErrorInStartBatch: a DoFn whose @StartBundle throws never reports its error, because SdkHarnessClient.newBundle sends the ProcessBundleRequest and then blocks in GrpcDataService.createOutboundAggregator waiting for a data stream that a bundle failing during setup never opens, so the run blocks for the data service's three-minute timeout. That is shared java-fn-execution behaviour; the Flink runner sickbays all of LifecycleTests and the Prism runner sickbays each of its three error tests. --- runners/kafka-streams/build.gradle | 10 +++++++ .../kafka/streams/TestKafkaStreamsRunner.java | 27 ++++++++++++++++++- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle index 52535c6654e6..2d0209eafbf5 100644 --- a/runners/kafka-streams/build.gradle +++ b/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 } diff --git a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunner.java b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunner.java index 1fe4b310789f..eb1ebf707746 100644 --- a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunner.java +++ b/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. * + *

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. + * *

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. + * + *

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;