[GSoC 2026] Kafka Streams runner: run on a real broker, correctly across partitions by junaiddshaukat · Pull Request #39546 · apache/beam

junaiddshaukat · 2026-07-29T09:47:57Z

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.

Adds what the runner needs to execute on an actual Kafka cluster, and an integration test that runs a pipeline through the production KafkaStreamsPipelineRunner against a Kafka container. Everything until now ran through TopologyTestDriver, which fakes the topics and never builds a KafkaStreams application, so the production path had not been executed. KafkaStreamsTopicManager creates the topics the runner names for itself - a bootstrap topic per Impulse and primitive Read, and a repartition topic per GroupByKey - with an AdminClient before the application starts. Kafka Streams treats them as user topics because they are declared with explicit names, so it does not create them and refuses to start when a source topic is missing; the broker's auto-creation is not a substitute, since it is often disabled and gives the topic the broker's default partition count. Only topics carrying one of the runner's prefixes are created, so a topic the user named is never created implicitly. Adds topicPartitions and topicReplicationFactor options. Fixes two bugs the test driver could not catch. Validation demanded jobEndpoint, which KafkaStreamsPipelineOptions inherits as required from PortablePipelineOptions but which is a client-side option: this code runs on the job server for an already-submitted pipeline, so it rejected every valid job. Only the options the runner reads are checked now, as Flink's equivalent PortablePipelineRunner does. Separately, the result object registers a KafkaStreams state listener in its constructor but was built after start(), and Kafka Streams only accepts a listener in the CREATED state, so every real startup threw and the listener behind waitUntilFinish and cancel was never installed; the result is now built before the application starts. The integration test needs Docker, so the default test task excludes it and a brokerIntegrationTest task runs it. Also adds PortableWindowingStrategyTest, checking that the windowing the runner reconstructs comes from the standard beam:window_fn URNs every SDK emits rather than anything Java-specific.

gemini-code-assist · 2026-07-29T09:48:00Z

Caution

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

github-actions · 2026-07-29T10:23:37Z

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

Every transform stamped its watermark reports as partition 0 of 1. That is right for a report handed to a fused child, which runs in the same task and sees exactly one instance of its parent, but wrong for a report crossing a repartition topic: the sink's partitioner broadcasts each report to every partition, so a downstream task sees one from every instance of the upstream transform. With all of them claiming to be the only partition it would treat the first report as if every instance had reported, advance its watermark, and fire before the other partitions had delivered their data. Identity is now 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 forwarding in process. The partition count a transform runs at is threaded through translation: one everywhere, raised only by a GroupByKey's shuffle, and inherited by everything fused downstream of it. Also pins the bootstrap topics to a single partition. An Impulse or a Read emits once per task, gated by a state store that is itself per task, so a multi-partition bootstrap topic would fire the same Impulse once per partition; only the repartition topic of a GroupByKey carries the pipeline's parallelism. Adds a broker test running two chained GroupByKeys across four partitions, which is the shape this affects — the second GroupByKey aggregates the reports of every task of the first, and a single-GroupByKey pipeline passes either way — plus a one-partition control and unit tests for the restamping.

je-ik

Great to see how the runner evolves towards a real usable runner! I have left a few questions.

je-ik · 2026-07-31T07:45:03Z

      lastForwardedWatermark = advanced;
-      // Stamped with this Flatten's own transform id; Flatten is a single instance for now, so the
-      // report is for its only partition (0 of 1).
+      // Stamped as the only source a consumer will see; a shuffle downstream restamps.


The "stamp" here resembles "timestamp", is this more a "marker"?

Good point, that reads wrong. Switched to "label"/"relabel" throughout, the shuffle relabels the report with the reporting instance's identity, which is what it actually does and doesn't collide with timestamps.

je-ik · 2026-07-31T07:46:11Z

      String parentProcessor = context.getProcessorNameForPCollection(inputPCollectionId);
      parentProcessors.add(parentProcessor);
      upstreamTransformIds.add(parentProcessor);
+      partitionCount = Math.max(partitionCount, context.getPartitionCount(inputPCollectionId));


Can getPartitionCount be zero or negative? Under which circumstances?

It could, and that was a gap. It returns 1 for anything unregistered, and otherwise whatever --internalParallelism was set to, which nothing validated. Worse than a bad topic: it's also the number of watermark reports a shuffle's consumer waits for, so a value below 1 would have left the consumer waiting forever rather than failing. The runner now rejects it before translating, and I documented the invariant where the count is read.

je-ik · 2026-07-31T07:48:21Z

+          + " of each Impulse and Read, and the repartition topic of each GroupByKey). This is the"
+          + " parallelism the shuffled parts of the pipeline can reach.")
+  @Default.Integer(1)
+  int getTopicPartitions();


This should be named so that it is clear from the command line where it will be used. --topicPartitions is not too clear, how about --internalParallelism?

Renamed, and I rewrote the description to say where it applies — a GroupByKey runs one task per partition of its repartition topic, so it's the number of instances its state and downstream stages spread over.

je-ik · 2026-07-31T07:49:50Z

+
+  @Description("Replication factor for the topics the runner creates for a pipeline.")
+  @Default.Short(1)
+  short getTopicReplicationFactor();


We should probably enable a complete set of topic-level configurations here. We can postpone it, but it would be good to create an issue. We might pass a JSON-serialized dictionary and/or publish the most-common options as separate methods.

Agreed, filed: #39565. Noted both ideas there — a JSON-serialized dictionary and/or the most common ones as separate options.

je-ik · 2026-07-31T07:55:05Z

+ * topology belongs to the user (a source or sink they named), and creating those implicitly would
+ * hide a misconfiguration behind an empty topic.
+ */
+class KafkaStreamsTopicManager {


I'm thinking if this should be implicit, or if we should add a life-cycle management:

# just create topics $ java -cp ... my.class --runner=KafkaStreams [... other opts ...] --create-topics # run $ java -cp ... my.class --runner=KafkaStreams [... other opts ...] # delete runner-created topics $ java -cp ... my.class --runner=KafkaStreams [... other opts ...] --delete-topics

But this is definitely out of scope of this PR.

Filed for later as you suggested: #39566, with your create/run/delete sketch. Agreed it's out of scope here.

je-ik · 2026-07-31T07:56:12Z

+ * replace running a pipeline from another SDK end to end, which additionally exercises the coders
+ * and the SDK harness; that needs the job server against a real broker.
+ */
+public class PortableWindowingStrategyTest {


The portable part here would be if we implemented own WindowFn, because that would have to go through the SDK harness. Otherwise the "classic" windowFns can be interpreted directly by the ReduceFnRunner.

You're right, and the test was overclaiming — I've renamed it from PortableWindowingStrategyTest to StandardWindowFnTranslationTest and rewritten the docs. What it actually pins down is that the standard WindowFns are rebuilt from the language-neutral URNs, so a pipeline from another SDK using fixed/sliding/session/global windows produces a strategy we can reconstruct. A user-written WindowFn is opaque to us and would have to run through the SDK harness that owns it, which we don't support — windowFnFromProto rejects the unrecognised URN. That's stated explicitly now.

je-ik · 2026-07-31T07:56:50Z

+
+  @BeforeClass
+  public static void startBroker() {
+    kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.1"));


Could we use official Apache image?

Switched. One wrinkle worth flagging: Testcontainers 1.21.4 can't bring up apache/kafka:3.9.0 — the container exits during startup, though the same image runs fine standalone. apache/kafka:4.0.0 works, so the test uses that while the runner's client stays on 3.9.0; a client talking to a newer broker is the direction Kafka supports. Comment in the code explains it.

Every transform reported its watermark as partition 0 of 1. That is right for a report handed to a fused child, which runs in the same task and sees exactly one instance of its parent, but wrong for a report crossing a repartition topic: the sink's partitioner broadcasts each report to every partition, so a downstream task sees one from every instance of the upstream transform. With all of them claiming to be the only partition it would treat the first report as if every instance had reported, advance its watermark, and fire before the other partitions had delivered their data. The identity is now attached where the report stops being delivered in process and starts crossing a topic: ShuffleByKeyProcessor relabels 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. The partition count a transform runs at is threaded through translation: one everywhere, raised only by a GroupByKey's shuffle, and inherited by everything fused downstream of it. Also pins the bootstrap topics to a single partition. An Impulse or a Read emits once per task, gated by a state store that is itself per task, so a multi-partition bootstrap topic would fire the same Impulse once per partition; only the repartition topic of a GroupByKey carries the pipeline's parallelism. Review changes: renames --topicPartitions to --internalParallelism, which says where it applies; rejects a value below one, since it is also the number of watermark reports a shuffle's consumer waits for and a non-positive value would leave it waiting rather than failing; drops the word stamp for label, which does not read like timestamp; and moves the broker test to the official apache/kafka image. Adds a broker test running two chained GroupByKeys across four partitions, which is the shape this affects, plus a one-partition control and unit tests for the relabelling. Renames PortableWindowingStrategyTest to StandardWindowFnTranslationTest and corrects what it claims: the standard WindowFns are interpreted runner-side by ReduceFnRunner, so the test pins down that they are rebuilt from the language-neutral URNs, and a WindowFn the user wrote themselves would instead have to run through the SDK harness, which the runner does not support.

github-actions Bot added the runners label Jul 29, 2026

github-actions Bot added the Next Action: Reviewers label Jul 29, 2026

junaiddshaukat changed the title ~~[GSoC 2026] Kafka Streams runner: run against a real broker~~ [GSoC 2026] Kafka Streams runner: run on a real broker, correctly across partitions Jul 30, 2026

je-ik self-requested a review July 31, 2026 07:42

je-ik reviewed Jul 31, 2026

View reviewed changes

This was referenced Jul 31, 2026

Kafka Streams runner: expose topic-level configuration for runner-created topics #39565

Open

Kafka Streams runner: lifecycle management for runner-created topics #39566

Open

Uh oh!

Conversation

junaiddshaukat commented Jul 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Creating the runner's topics

Watermark identity across a shuffle

Two bugs the test driver could not catch

Testing

Uh oh!

gemini-code-assist Bot commented Jul 29, 2026

Uh oh!

github-actions Bot commented Jul 29, 2026

Uh oh!

je-ik left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

junaiddshaukat commented Jul 29, 2026 •

edited

Loading