[GSoC 2026] Kafka Streams runner: bound a bundle by element count

[junaiddshaukat]

Summary

Part of #18479.

A bundle stayed open until the next watermark arrived. On a stream that produces steadily that means it grows without limit, and on one where watermarks are sparse the elements already fed to it are not emitted for as long as the gap lasts. maxBundleSize was declared as a pipeline option but nothing read it.

The stage now counts the elements fed to the open bundle and closes it once the bound is reached.

Commits at bundle boundaries

Closing a bundle asks Kafka Streams to commit, so the elements a bundle consumed and the records it produced are committed together: a restart replays either the whole bundle or none of it.

Worth being precise about what that does and does not give. It aligns commits to bundle boundaries; it does not prevent Kafka Streams from committing on its own interval part-way through a bundle. Ruling that out would mean closing the bundle from a pre-commit hook, so that the bundle is always finished before offsets are committed. The class documents this.

maxBundleTimeMs is still not applied

The natural implementation — close the bundle from a wall-clock punctuator once it has been open longer than the bound — does not work against a real broker, so it is not in this PR and the option documents that it currently has no effect.

With the punctuator in, the integration test that runs two chained GroupByKeys across four partitions emits its single group about six times, and the count keeps climbing after the input has stopped. Without it, the count is exactly one.

What that leaves out:

• Not the metrics folding. The runner folds each bundle's metrics into the job's step map by adding them, so more bundles could in principle inflate a counter. MetricsAcrossBundlesTest splits the same input across many bundles and the counter is unchanged, so per-bundle reports are per-bundle.
• Not ProcessorContext.commit(). With the punctuator disabled but the commit still requested on every close, the test passes. With the punctuator enabled but the commit request removed from it, the duplication still happens.

So it appears to be closing a Fn-API bundle from a punctuator rather than from record processing, and I would rather leave a documented gap than ship behaviour whose failure mode I cannot explain. A pre-commit hook — closing the bundle from a state store's flush() — looks like the shape that would fix both this and the mid-bundle commit noted above, and I would like to agree the approach before implementing it.

Testing

• BundleBoundaryTest — a bound the input passes several times over produces several bundles; a bound the input never reaches leaves a single one. The elements have to arrive as separate records for the bound to count them, so the pipelines read from a Create rather than fanning out inside one stage, which fusion would collapse into a single input.
• MetricsAcrossBundlesTest — a user counter is unchanged when the same input is split across many bundles.

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

[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: @jrmccluskey 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).