You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@beam.apache.org by "junaiddshaukat (via GitHub)" <gi...@apache.org> on 2026/08/01 15:30:38 UTC

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

junaiddshaukat opened a new pull request, #39578:
URL: https://github.com/apache/beam/pull/39578

## 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
```

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org