[GSoC 2026] Kafka Streams runner: run on a real broker, correctly across partitions

[junaiddshaukat]

Summary

Part of #18479.

Makes the runner able to run on a real Kafka cluster, and makes it correct when a pipeline is spread across more than one partition. 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 and the distributed path had never been exercised at all. Running both found four bugs.

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. The broker's auto.create.topics.enable is not a substitute — 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, and only creates topics carrying one of the runner's own prefixes, so a topic the user named is never created implicitly. Bootstrap topics are always created with a single partition: an Impulse or a Read emits once per task, gated by a state store that is itself per task, so a bootstrap topic with several partitions would fire the same Impulse once per partition. The repartition topic of a GroupByKey is what carries the pipeline's parallelism, and takes the new topicPartitions option (topicReplicationFactor comes with it).

Watermark identity across a shuffle

Every transform stamped its watermark reports as partition 0 of 1. That is right for a report handed to a fused child — it runs in the same task and sees exactly one instance of its parent — but wrong for a report that crosses a repartition topic, because the sink's partitioner broadcasts each report to every partition. A downstream task therefore sees a report from every instance of the upstream transform, and with all of them claiming to be the only partition it would treat the first report as if every instance had already reported, advance its watermark, and fire before the other partitions had delivered their data.

So identity is attached where the report stops being delivered in process and starts crossing a topic: ShuffleByKeyProcessor restamps it. That processor runs in the upstream transform's own task, so taskId().partition() is exactly the instance the report is for, and the transform id is left alone so a downstream aggregator still matches the report to the upstream it expects. Transforms go on reporting a single source when they forward in process.

The partition count a transform runs at is threaded through translation: it is one everywhere, only a GroupByKey's shuffle raises it, and everything fused downstream of that inherits it.

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, and Flink's equivalent PortablePipelineRunner does not validate here either. Only the options the runner reads are checked now.

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 was built after start(), and Kafka Streams only accepts a listener while the application is in CREATED. This threw on every real startup, and that listener is what backs waitUntilFinish() and cancel(). The result is now built before the application starts, and the requirement is documented on its constructor.

Testing

KafkaStreamsRunnerBrokerIT runs pipelines through the production KafkaStreamsPipelineRunner against a Kafka container, polling the pipeline's metrics for the expected output:

• a GroupByKey on one partition, covering topic creation, a real repartition-topic round trip, exactly-once, the changelog and the application lifecycle;
• two chained GroupByKeys on one partition;
• the same two chained GroupByKeys across four partitions, which is the shape that needs the watermark identity above — the second GroupByKey aggregates the reports of every task of the first. A single-GroupByKey pipeline passes either way, which is why the chained one is the test that matters.

ShuffleByKeyProcessorTest covers the restamping directly: the instance identity attached, two instances staying distinct, an unpartitioned upstream still reporting a single source, and data records passing through untouched.

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 the default test task excludes it and a separate task runs it:

./gradlew :runners:kafka-streams:brokerIntegrationTest   # 3 tests
./gradlew :runners:kafka-streams:validatesRunner         # 49 tests
./gradlew :runners:kafka-streams:build                   # 76 unit tests, spotless + checker + errorprone

Also adds PortableWindowingStrategyTest, which checks that the windowing the runner reconstructs comes from the language-neutral beam:window_fn:* URNs every SDK emits rather than the Java-serialized form — a question raised on the windowing PR about pipelines 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).