ConsumerSupervisor

The supervisor's whole mutable state, as one immutable value behind one reference.

One AtomicReference to a case class rather than five references: a reader of ConsumerSupervisor.status gets a consistent view, where five separate atomics would let it observe state = running alongside the previous generation's error.

The longest one ConsumerSupervisor.status can honestly take, for a caller that has to await it.

A budget has to be able to contain the calls it wraps. The probe's was lag.requestTimeout — the bound on a single admin call — while status makes StatusRoundTrips of them one after another. Under a slow broker the await therefore expired on every tick before the call it wrapped could return, and consume.running and consume.checkpoint.divergence froze: a paused consumer became indistinguishable from a crashed one exactly while somebody was looking at the dashboard.

max and not +, because the two reads run concurrently. The checkpoint query and the broker round trips ask different systems and neither answer feeds the other, so ConsumerSupervisor.status starts both and awaits both. Summing them would price a wait that does not happen — and the sum was what pushed this budget past the lag poller's own interval, giving a slow tick time to overlap the next one.

StoreTimeout is the allowance for the checkpoint read: the bound the seek path puts on the same query, borrowed here because the database offers no other number.

How long a drain may take before the supervisor gives up waiting and reports the consumer stopped anyway.

Longer than ConsumerConfig.drainTimeout, so the connector's own bound is the one that normally fires and this is only the backstop for a drain that hangs entirely.