[GSoC 2026] Kafka Streams runner: TestPipeline-dispatchable test runner (PAssert works)#39362
Conversation
Adds TestKafkaStreamsRunner, a PipelineRunner that TestPipeline-based suites -- including Beam's @ValidatesRunner tests -- can dispatch to. It fills in the options the translation needs (applicationId, EMBEDDED harness), drives the pipeline to quiescence through TopologyTestDriver via KafkaStreamsTestRunner, and returns a terminal PipelineResult exposing the harness-reported metrics -- which TestPipeline.verifyPAssertsSucceeded uses to confirm every PAssert in the pipeline ran and succeeded. Tests prove the full PAssert mechanism against the runner: containsInAnyOrder materializes the actual PCollection through the GBK + Flatten bootstrap path (no side inputs), an assertion on Count.perElement covers the Combine-by- expansion case, and a failing assertion fails the run with the assertion text (over the Fn API a DoFn error travels as an error string, so the metrics-based count check is what catches silently dropped assertions). Next: the validatesRunner gradle task enabling Beam's suite with a sickbay list.
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request adds a specialized test runner for the Kafka Streams implementation in Apache Beam. By providing a dispatchable runner that leverages the TopologyTestDriver and an embedded harness, it enables robust testing of pipelines and PAssert assertions. This is a critical step toward fully supporting the @ValidatesRunner test suite for the Kafka Streams runner. Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces TestKafkaStreamsRunner and its associated unit tests to allow running TestPipeline-based suites against the Kafka Streams runner without requiring an active broker. The feedback suggests two improvements: first, enhancing the applicationId check in TestKafkaStreamsRunner to handle empty strings in addition to null values; second, updating the exception assertion in TestKafkaStreamsRunnerTest from RuntimeException to Throwable to robustly catch potential AssertionError occurrences.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if (options.getApplicationId() == null) { | ||
| options.setApplicationId("ks-validates-runner-" + UUID.randomUUID()); | ||
| } |
There was a problem hiding this comment.
It is safer to check if the applicationId is either null or empty. If an empty string is provided as the application ID, it will bypass the null check but will subsequently fail in Kafka Streams, which requires a non-empty application ID.
| if (options.getApplicationId() == null) { | |
| options.setApplicationId("ks-validates-runner-" + UUID.randomUUID()); | |
| } | |
| if (options.getApplicationId() == null || options.getApplicationId().isEmpty()) { | |
| options.setApplicationId("ks-validates-runner-" + UUID.randomUUID()); | |
| } |
There was a problem hiding this comment.
Done, also guards the empty string.
| RuntimeException thrown = assertThrows(RuntimeException.class, pipeline::run); | ||
| assertThat(Throwables.getStackTraceAsString(thrown), containsString("AssertionError")); |
There was a problem hiding this comment.
Using assertThrows(RuntimeException.class, ...) can cause the test to fail if the runner's unwrapping logic successfully extracts and rethrows an AssertionError (since AssertionError extends Error and is not a subclass of RuntimeException). To make the test robust and correctly handle both wrapped RuntimeException and unwrapped AssertionError, assert that a Throwable is thrown instead.
| RuntimeException thrown = assertThrows(RuntimeException.class, pipeline::run); | |
| assertThat(Throwables.getStackTraceAsString(thrown), containsString("AssertionError")); | |
| PAssert.that(pipeline.apply("create", Create.of(1, 2, 3))).containsInAnyOrder(1, 2, 4); | |
| Throwable thrown = assertThrows(Throwable.class, pipeline::run); |
There was a problem hiding this comment.
Good point, if the chain ever preserves the DoFn's AssertionError the runner rethrows it as an Error, which would have failed this test. Switched to Throwable with the message check.
87911f0
into
apache:feat/18479-kafka-streams-runner-skeleton
2 participants
Summary
Adds
TestKafkaStreamsRunner, aPipelineRunnerthatTestPipeline-based suites -- including Beam's@ValidatesRunnertests -- can dispatch to via--runner. Part of #18479, second step toward the @ValidatesRunner suite (after the metrics PR #39341).applicationId, theEMBEDDED harness) on the pipeline's options, translates and drives the
topology to quiescence through
TopologyTestDriver, and returns a terminalPipelineResultexposing the harness metrics.TestPipeline.run()verifies through those metrics that everyPAssertactually ran and succeeded (
verifyPAssertsSucceededcountsPAssert.SUCCESS_COUNTER) -- this is what the metrics PR was for.PAssertfails the run with the assertion text in the error. Overthe Fn API a DoFn failure travels as an error string rather than a Java
AssertionErrorobject, so the metrics count check is what catches silentlydropped assertions (e.g. a watermark bug suppressing the assertion DoFn).
Tests:
containsInAnyOrderonCreate(PAssert's GBK + Flatten bootstrap materialization, no side inputs), an assertion onCount.perElement()(Combine.perKey works today via its GBK + ParDo expansion in the harness), andthe failing-assertion case.
Next up: the
validatesRunnergradle task enabling Beam's suite (with a sickbay list), following the FlinkcreateValidatesRunnerTaskpattern.