junaiddshaukat opened a new pull request, #39748: URL: https://github.com/apache/beam/pull/39748
## Summary Part of #18479. Two pipeline options that came out of trying to measure how the runner behaves under load, and that are worth having whatever the measurement ends up looking like. ## How much a source reads is not how big a bundle is `ReadTranslator` took the unbounded source's per-poll limit from `--maxBundleSize`. They are different concerns: a small bundle is how output arrives promptly, while how much a source reads at a time is about throughput. Sharing one setting means a pipeline cannot ask for both — lowering the bundle size to get prompt output also throttles the source to a couple of elements per poll. `--readMaxElementsPerPoll` (default 1000, what the source effectively had before) separates them. ## The recovery knob was unreachable How quickly work moves to another instance after one is lost is bounded by how quickly the consumer group notices, which is `session.timeout.ms`. The runner never set it, so it was Kafka's default of 45 seconds and a pipeline had no way to ask for anything else. `--sessionTimeoutMs` exposes it, keeping Kafka's 45s default so nothing changes for an existing pipeline. Lowering it trades tolerance of a slow or briefly paused instance for quicker recovery, and a broker will refuse a value below its own `group.min.session.timeout.ms`, which itself defaults to 6s — so this is not a knob that can be turned arbitrarily far. The heartbeat is derived as a third of the timeout rather than exposed separately. Kafka rejects a heartbeat that is not shorter than the session timeout, so deriving it keeps the pair consistent however the timeout is set; a third is the ratio Kafka's own defaults use. ## Testing ``` ./gradlew :runners:kafka-streams:build # 103 unit tests, spotless + checker + errorprone ./gradlew :runners:kafka-streams:validatesRunner # 59 tests ``` `KafkaStreamsPipelineRunnerConfigTest` covers the default, the override, the ratio, and that the heartbeat stays shorter than the timeout across a range of values. That last one is not decoration: the derivation first had a fixed 200ms floor, which produces a heartbeat equal to the timeout at 200ms and would be rejected by Kafka. The test fails if that floor is put back. -- 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]
