[GSoC 2026] Kafka Streams runner: enable ParDoTest in the ValidatesRunner suite

[junaiddshaukat]

Summary

Part of #18479.

Enables Beam's ParDoTest suites in the Kafka Streams runner's validatesRunner task, which takes the suite from 26 to 44 tests.

What runs now

Suite	Tests
ParDoTest$BasicTests	6
ParDoTest$MultipleInputsAndOutputTests	5
ParDoTest$TimestampTests	4
ParDoTest$LifecycleTests	3

MultipleInputsAndOutputTests passes on the multi-output executable stages that landed in #39410. The remaining ParDoTest nested classes (state, timers, timer families, bundle finalization, and the side-input half of MultipleInputsAndOutputTests) fall out on category excludes the task already declares, so no new excludes were needed.

Surfacing a failed run as PipelineExecutionException

The three LifecycleTests error tests assert that a DoFn throwing surfaces as a RuntimeException carrying the user's message. What came out of TestKafkaStreamsRunner.run instead was the raw Kafka Streams wrapper:

java.lang.AssertionError:
Expected: (an instance of java.lang.RuntimeException and exception with message a string containing "test error in process")
     but: ... message was "stream-thread [Time-limited test] task [0_0] Exception caught while punctuating processor 'Create-Values-Read-CreateSource-'"

The user's exception is at the bottom of that chain, under several layers of bundle- and Fn-API-level wrappers. run now restates a failed run as Pipeline.PipelineExecutionException carrying the root cause, so the user's message ends up on the thrown exception — RuntimeException(Throwable) derives its message from the cause. SparkPipelineResult and DirectRunner restate failures the same way. The existing unwrap that rethrows an AssertionError (a failed PAssert) still runs first.

That fixes testParDoWithErrorInProcessElement and testParDoWithErrorInFinishBatch.

One sickbay

testParDoWithErrorInStartBatch is sickbayed. 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. 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.

Testing

./gradlew :runners:kafka-streams:validatesRunner   # 44 tests, 0 failures
./gradlew :runners:kafka-streams:build             # 61 unit tests, spotless + checker clean

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[github-actions]

Assigning reviewers:

R: @chamikaramj added as fallback since no labels match configuration

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).