[GSoC 2026] Kafka Streams runner: bound a bundle by element count - #39578
[GSoC 2026] Kafka Streams runner: bound a bundle by element count#39578junaiddshaukat wants to merge 2 commits into
Conversation
A bundle stayed open until the next watermark, so on a stream that produces steadily it grew without limit and nothing it had already processed was emitted until a watermark happened to arrive. 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 that many have gone in, and closing a bundle asks Kafka Streams to commit, so the elements a bundle consumed and the records it produced are committed together and a restart replays either all of the bundle or none of it. That aligns commits to bundle boundaries; it does not stop Kafka Streams from committing on its own interval part-way through a bundle, which would need a pre-commit hook, and the class documents that. maxBundleTimeMs is still not applied. Closing a bundle from a wall-clock punctuator made the pipeline with two chained GroupByKeys across four partitions emit its group repeatedly against a real broker, with the count still climbing after the input stopped. The same bundles closed from the record path are fine, and requesting the commit is not the cause — the duplication happens with the commit request removed. Rather than ship behaviour whose failure mode is not understood, the option documents that it has no effect yet. BundleBoundaryTest covers the size bound and the case of a bound the input never reaches. MetricsAcrossBundlesTest pins down that splitting the same input across many bundles does not change a user counter, since the runner folds each bundle's metrics by adding them.
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Assigning reviewers: R: @jrmccluskey added as fallback since no labels match configuration Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
|
|
||
| private void closeBundleAndFlush(Record<byte[], KStreamsPayload<?>> record) { | ||
| byte[] key = record.key(); | ||
| closeBundleAndFlush(key == null ? lastKey : key, record.timestamp()); |
There was a problem hiding this comment.
This feels weird. If the key is null, we should not inject the "last" key, because it might be wrong. Does this mean we have "unkeyed" (i.e. stateless) ExecutableStage?
There was a problem hiding this comment.
You're right, and it's worse than unnecessary standing in a different record's key there would just be wrong.
That parameter existed only because I had the bundle also being closed by a wall-clock punctuator, which has no record to take a key from, so it passed the last key it had seen. I dropped the punctuator from this PR when it turned out to duplicate output, and I left the substitution behind. It's gone now — the close takes the record again and carries that record's own key and timestamp, exactly as before this PR.
And yes, to your question: an ExecutableStage is unkeyed. It runs stateless, with StateRequestHandler.unsupported() and no timer receivers, so the Kafka record key means nothing to it and is only carried along. Where the key does matter, downstream sets it — ShuffleByKeyProcessor derives it from the Beam key before a GroupByKey. I've added that to the javadoc so the next reader doesn't have to ask.
The bundle close took a key and a timestamp separately so that a wall-clock punctuator, which has no record of its own, could pass the last key it had seen. That punctuator is not in this PR, so the substitution only ever applied to a record whose own key was null — and standing in a different record's key there would be wrong rather than merely unnecessary. The close now takes the record again and carries that record's own key and timestamp onto the outputs, as it did before this PR, and the field holding the last key is gone. Documents why the key is incidental here: an executable stage is unkeyed, running stateless with no state or timers, so the Kafka record key is only carried along, and where it matters downstream sets it -- ShuffleByKeyProcessor derives it from the Beam key before a GroupByKey.
2 participants
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.
maxBundleSizewas 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.
maxBundleTimeMsis still not appliedThe 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:
MetricsAcrossBundlesTestsplits the same input across many bundles and the counter is unchanged, so per-bundle reports are per-bundle.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 aCreaterather 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.