junaiddshaukat opened a new pull request, #39745: URL: https://github.com/apache/beam/pull/39745
## Summary Part of #18479. Measures how long work takes to move to another instance when the instance doing it goes away — the claim the runner is proposed on, which up to now has been an argument from the architecture rather than a number. ## What it measures, and why it is built this way Each instance stamps what it processes with its own name. That is not decoration: the runner reads an unbounded source with a single reader (#39626), so at any moment exactly one instance is doing the work, and removing the other one proves nothing while producing a very flattering number. The test finds the instance that is actually producing, removes that one, and times how long until the other produces. It fails loudly if both are producing, since then the removal would not isolate a handover. There is no `GroupByKey` in the pipeline, deliberately. A windowed group emits only when a window fires, and the bundle downstream of it closes only on element count or on a watermark arriving (#39633), so its output comes in bursts about **23 seconds** apart — far coarser than the thing being measured. Reading and counting gives a steady signal, and the task reading the source is what has to move. ## Results | | Handover | Runs | |---|---|---| | As the runner is configured today | **~44 s** | 43.9 s, 44.3 s | | With the consumer set to leave the group on close | **~1.8 s** | 1.82 s, 1.82 s, 1.77 s | Kafka Streams does not send a `LeaveGroup` when it closes, so the group coordinator only notices the instance has gone once `session.timeout.ms` expires — 45 seconds by default, which is what the first row measures. **This PR changes no configuration.** Setting the consumer to leave the group on close makes scale-down 24x faster, but `internal.leave.group.on.close` is an internal Kafka config with no compatibility guarantee, and Kafka Streams defaults it off deliberately so that rolling restarts do not cause rebalance churn. For a runner whose instances are added and removed on purpose it looks like the right trade, but that is a decision for review rather than something to slip in with a test. ## What this does not measure The instance removed here shuts down in an orderly way, and this pipeline holds no state. A machine that dies does neither. It cannot announce its own departure, so no configuration makes it detectable faster than `session.timeout.ms`; and a pipeline with state has to restore that state before it can carry on. So these numbers are the cost of moving stateless work between instances, and a lower bound on anything else. Recovery from a crash is detection plus at least this. That is worth being plain about, because it bears on how the runner should be described: the advantage over a checkpoint-based engine is in what happens *after* a failure is detected — reassigning partitions and restoring from a changelog, rather than restarting the job graph from a checkpoint — and not in detecting the failure sooner. ## Testing ``` ./gradlew :runners:kafka-streams:build # 99 unit tests, spotless + checker + errorprone ./gradlew :runners:kafka-streams:validatesRunner # 59 tests ./gradlew :runners:kafka-streams:brokerIntegrationTest # 6 tests, including this one ``` The integration task now shows standard output, so the measurement is visible when the test is run; it is a manually invoked verification task, so the extra output costs nothing in CI. -- 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]
