This is an automated email from the ASF dual-hosted git repository.
je-ik pushed a commit to branch feat/18479-kafka-streams-runner-skeleton
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to
refs/heads/feat/18479-kafka-streams-runner-skeleton by this push:
new b615ae85aed [GSoC 2026] Kafka Streams runner: enable ParDoTest in the
ValidatesRunner suite (#39451)
b615ae85aed is described below
commit b615ae85aedfce799a94e4dc76152c13adc7f392
Author: M Junaid Shaukat <[email protected]>
AuthorDate: Thu Jul 23 13:36:14 2026 +0500
[GSoC 2026] Kafka Streams runner: enable ParDoTest in the ValidatesRunner
suite (#39451)
---
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 52535c6654e..2d0209eafbf 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 1fe4b310789..eb1ebf70774 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 @@ package org.apache.beam.runners.kafka.streams;
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 @@ import org.joda.time.Duration;
* 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 final class TestKafkaStreamsRunner extends
PipelineRunner<PipelineResult>
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;
