[GSoC 2026] Kafka Streams runner: run against a real broker

[junaiddshaukat]

Summary

Part of #18479.

Makes the runner able to run against a real Kafka cluster, and adds the integration test that proves it. Until now everything ran through TopologyTestDriver, which fakes the topics and never builds a KafkaStreams application — so the production path had never actually been executed. Running it found two bugs that the test driver cannot structurally catch.

This is also the groundwork for two things that were going to need it anyway: running the ValidatesRunner suite from another SDK through the job server, and the failure-recovery and scaling benchmark.

Creating the runner's topics

The runner shuffles through topics it names itself — a bootstrap topic per Impulse and per primitive Read, and a repartition topic per GroupByKey. Kafka Streams creates the internal topics it manages, but these are declared with explicit names through addSource and addSink, so to Kafka Streams they are ordinary user topics: it does not create them, and refuses to start with MissingSourceTopicException when a source topic is missing. Leaving it to the broker's auto.create.topics.enable is not a fix either — it is off on many clusters, and a topic auto-created on first fetch gets the broker's default partition count instead of the pipeline's.

KafkaStreamsTopicManager creates them with an AdminClient before the application starts. It only creates topics carrying one of the runner's own prefixes; anything else in the topology is a topic the user named, and creating those implicitly would hide a misconfiguration behind an empty topic. A topic that appears between the check and the create — another instance of the same job starting concurrently — is treated as success.

Two new options come with it: topicPartitions (the parallelism the shuffled parts of a pipeline can reach) and topicReplicationFactor.

Two bugs the test driver could not catch

Options validation rejected every valid job. run validated the whole KafkaStreamsPipelineOptions interface, which inherits @Required jobEndpoint from PortablePipelineOptions:

java.lang.IllegalArgumentException: Missing required value for [...getJobEndpoint(), "Job service endpoint to use..."]

jobEndpoint is a client-side option — the address a client uses to reach the job server. This code runs on the job server, executing a pipeline that has already been submitted, so it does not apply. Flink's equivalent PortablePipelineRunner does not validate here either. Now only the options this runner actually reads are checked, which keeps the clear error that motivated the original call.

The state listener was registered after the application had started.

java.lang.IllegalStateException: Can only set StateListener before calling start(). Current state is: REBALANCING

The result object registers a KafkaStreams state listener in its constructor, but it was constructed after start() had been called, and Kafka Streams only accepts a listener while the application is still in CREATED. This threw on every real startup, and the listener is what backs waitUntilFinish() and cancel(). The result is now built before the application starts, and the ordering requirement is documented on its constructor.

Testing

KafkaStreamsRunnerBrokerIT runs Impulse -> ParDo -> GroupByKey -> ParDo through the production KafkaStreamsPipelineRunner against a Kafka container, and polls the pipeline's metrics until the group counter reaches the expected value. That covers the runner creating its own topics, records round-tripping through a real repartition topic, exactly-once processing, the state stores' changelog, and the application lifecycle.

It uses testcontainers rather than an embedded broker because library.java.testcontainers_kafka is already a declared Beam dependency and is what KafkaIO's own tests use.

It needs Docker, so it is not part of the default build — the default test task excludes it and a brokerIntegrationTest task runs it:

./gradlew :runners:kafka-streams:brokerIntegrationTest   # 1 test, passes in ~9s once the container is up
./gradlew :runners:kafka-streams:validatesRunner         # 49 tests
./gradlew :runners:kafka-streams:build                   # 72 unit tests, spotless + checker + errorprone

Also adds PortableWindowingStrategyTest, which checks that the windowing the runner reconstructs comes from the language-neutral windowing strategy in the proto — the standard beam:window_fn:* URNs every SDK emits — rather than from anything Java-specific. That was a question raised on the windowing PR about whether the runner would work for a pipeline submitted from another SDK.

[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).