[GSoC 2026] Kafka Streams runner: terminate a bounded pipeline when it is drained

[junaiddshaukat]

Summary

Part of #18479. A bounded pipeline never finished against a real broker: Kafka Streams runs a topology until something closes the client, and nothing did. The pipeline produced the right answer and then sat there, and waitUntilFinish() blocked for ever.

This was invisible to every existing suite. The @ValidatesRunner tests run on TopologyTestDriver, which is synchronous and reports DONE unconditionally; the broker integration tests poll a counter and then cancel(), so none of them ever waits for a pipeline to finish on its own.

How termination is decided

Every processor already emits TIMESTAMP_MAX_VALUE once its input is exhausted, so the runner knows when it is drained — it just did nothing with it. Each processor now schedules its own termination when it emits that watermark, and the client is closed once they have all reported.

Scheduling rather than reporting inline is what makes it safe: a punctuator runs after the current record has been handled, so flushing a bundle, forwarding downstream and committing all still happen first. It has to be WALL_CLOCK_TIME, since no records arrive after the final watermark and stream time would never advance. The interval is 1ms because Kafka Streams rejects anything smaller.

Two things that turned out to matter:

It waits for every processor, not the first. One instance can own tasks from both sides of a repartition topic. The upstream side goes terminal as soon as it has written to the topic while the downstream side still has to consume it, so stopping at the first would discard that work and report a successful run.

It waits until the topology has finished starting. Processors register as their task is initialized, so mid-startup the registered set is only part of the pipeline. On a short pipeline the source can drain before the stage downstream of the repartition topic exists, and stopping there reports success having produced nothing.

No coordination between instances is needed. Watermarks crossing a repartition topic are broadcast to every partition, so every task observes the terminal watermark wherever it runs, and each instance reaches the same conclusion on its own.

run() now blocks

JobInvocation reads the pipeline result's state once, when run() returns. Ours returned right after start(), so a job stayed RUNNING for ever even after the client had stopped cleanly. It now blocks until the pipeline finishes, which is what FlinkPipelineRunner does by blocking in executor.execute().

This is a deliberate contract change: an unbounded pipeline blocks the calling thread until the job is cancelled. That is the same behaviour Flink has, and the job service already runs run() on its own executor and interrupts it to cancel — which the runner now handles by closing the client when it sees the interrupt.

An unrelated bug this turned up

Repartition topics were named __beam_gbk_<transformId>, with no application id, while the Impulse and Read bootstrap topics already include one. Transform ids come from the pipeline's structure, so two jobs running the same pipeline shuffled through the same topic and read each other's data. Because the topic is only created when it does not already exist, the second job also silently inherited the first one's partition count instead of the one it asked for.

It surfaced here because the new integration test runs the same pipeline at a different parallelism to an existing one, which no two tests had done before. It is a small fix and independent of the rest of this PR — happy to split it out if you would rather review it separately.

Note that this renames the topics, so a job restarted after this change shuffles through new ones and leaves its old repartition topics behind. That seems fine for a runner that is on a feature branch and marked experimental, but it is a behaviour change rather than a pure fix.

Testing

./gradlew :runners:kafka-streams:build                    # 95 unit tests, spotless + checker + errorprone
./gradlew :runners:kafka-streams:validatesRunner          # 59 tests
./gradlew :runners:kafka-streams:brokerIntegrationTest    # 4 tests

TerminationTrackerTest covers the decision logic, including the two cases above. I checked each fails without the code that makes it pass, rather than trusting that they were green.

KafkaStreamsRunnerBrokerIT.aBoundedPipelineTerminatesOnItsOwn is the end-to-end one: two chained GroupByKeys across four partitions, and nothing cancels it, so returning from run() at all is the assertion. It also asserts the output is produced exactly once, because termination rides a wall-clock punctuator and that is the same mechanism that duplicates output when it is used to close bundles on time (#39633).

The broker suite also got faster — 2m56s to 31s — since the topic fix stopped the tests inheriting each other's partition counts.

Beyond the suites, a Beam Python pipeline now runs and finishes on its own against a real broker: Create -> Map, terminating in about ten seconds, reporting DONE, with the expected output. Before this it hung indefinitely.

[github-actions]

Assigning reviewers:

R: @kennknowles 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).

[je-ik]

One minor comment, this can be merged.

[je-ik]

Should we nullify the callback here so that it cannot be run twice?

[junaiddshaukat]

Good call. It couldn't actually run twice before — there was a fired flag guarding it — but nulling the callback does the same job, so I've dropped the flag and kept just the one mechanism.

It's a real path rather than a theoretical one: the callback stops the client, which closes every task's processors, and each of those unregisters on the way out and asks the tracker again. Added a test for that, and checked it fails if the callback isn't cleared.