junaiddshaukat opened a new pull request, #39611: URL: https://github.com/apache/beam/pull/39611
## Summary Part of #18479. The runner could only read sources that finish. That is the wrong shape for what it is: a Kafka Streams application is a long-running stream processor, and a pipeline over bounded data has more efficient homes. This adds `UnboundedSource` support, so the runner can read a source that never ends. ## How it differs from the bounded read Bounded and unbounded reads share the `beam:transform:read:v1` URN and are told apart by `ReadPayload.getIsBounded()`, so `ReadTranslator` now branches there. What sits behind the branch is genuinely different: - **The source is polled, not drained.** `advance()` returning false means nothing is available *right now*, not that the source is finished, so the reader is asked again on each turn of a wall-clock punctuator. A poll takes at most `--maxBundleSize` elements, so a fast source cannot monopolise the Kafka Streams thread and starve the rest of the topology. - **The watermark comes from the reader.** A bounded read jumps to the end of time once its input runs out. Here `UnboundedReader#getWatermark()` is forwarded whenever it advances, which is what lets downstream windows close on a stream that never finishes. ## Resuming after a restart `UnboundedReader#getCheckpointMark()` describes the position the reader has consumed to. It is written to a persistent state store, and the reader is created from the stored mark rather than from scratch, so a task that restarts or moves resumes where it left off instead of re-reading from the beginning. The mark is written *after* the elements it covers have been forwarded, so it can never claim more progress than was actually emitted. The store is changelogged and, under exactly-once, its writes commit atomically with the records the processor forwarded. ## What this does not do yet - **`finalizeCheckpoint()` is not called.** A mark should be finalized once it is durably committed, which needs a pre-commit hook the runner does not have — the same gap that stopped the bundle time bound in #39578. Sources that rely on finalization to acknowledge or release data will not see it. Worth doing as its own change, once that hook exists. - **No split distribution.** The source is read by a single reader, so a source with several splits is consumed by one instance. Spreading splits across instances belongs with the topic-based shuffle work. ## Testing `UnboundedReadTest` runs a pipeline over `CountingSource.unbounded()`. Nothing caps the source — capping it with `withMaxNumRecords` would turn it back into a bounded read and test the wrong path — so the work is bounded by the per-poll element limit and the number of turns the test drives. It asserts more than a single poll's worth of elements arrive, which is the property that separates this from a bounded read: the source has to be asked again on each turn and carry on from where it was. The elements are also checked to be contiguous from zero, so a poll neither skips nor repeats what the previous one consumed. ``` ./gradlew :runners:kafka-streams:validatesRunner # 59 tests, 0 failures ./gradlew :runners:kafka-streams:build # 84 unit tests, spotless + checker + errorprone ``` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
