junaiddshaukat opened a new pull request, #39761: URL: https://github.com/apache/beam/pull/39761
## Summary Part of #18479. An unbounded source could hold the Kafka Streams thread indefinitely, and a pipeline that did so read steadily and produced no output at all. ## What happens `UnboundedReadProcessor` polls its source from a wall-clock punctuator scheduled every 50ms, and the same thread runs the rest of the topology. Each turn was bounded by a count — `--readMaxElementsPerPoll` per batch, `--readCheckpointNumBundles` batches — and a count cannot bound the time, because how long an element takes is decided by the pipeline underneath the source rather than by the source. Once a turn takes longer than the interval it is scheduled at, it is already due again when it returns and fires straight away. The source then keeps the thread and the stages below it are never scheduled. Measured with a grouping pipeline over 2000 keys: | `--readMaxElementsPerPoll` | time per poll | share of the thread | groups produced | | --- | --- | --- | --- | | 200 | 3ms | 6% | 52,000 | | 5000 | 57ms | 89% | none | In the second case the pipeline read 40M elements and emitted nothing, and the grouping's consumer had read 682 records out of 11.6M. It is not a pipeline falling behind — it is one that never runs the rest of itself. ## The change `--readMaxPollTimeMs` (default 10) bounds a turn in time as well as in count; whichever bound is reached first ends it. The turn's deadline is checked between batches and, since one batch can be long enough on its own, every 64 elements within a batch — often enough to bound the overshoot, not so often as to put a clock read in front of every element. Roughly, the source takes `readMaxPollTimeMs` of each 50ms interval, so the default leaves the thread four fifths of its time. Lowering it yields sooner and reads less per turn; raising it does the reverse, and above 50ms it re-creates the behaviour above. At `--readMaxElementsPerPoll=5000`, which produced no groups at all before, the pipeline now produces every window complete. ## Testing ``` ./gradlew :runners:kafka-streams:build # unit tests, spotless + checker + errorprone ./gradlew :runners:kafka-streams:validatesRunner # 59 tests ``` Two tests in `UnboundedReadTest`. `aPollOutOfTimeYieldsBeforeReachingItsElementBound` gives a turn no time at all, so what stops it can only be the time bound, and asserts it yields short of its element bound; `aPollWithTimeToSpareReachesItsElementBound` is the control, confirming the time bound only ever cuts a turn short. The first fails if the deadline is pushed out of reach, checked with a mutation that still compiles — a mutation that fails to compile proves nothing. -- 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]
