unknowntpo opened a new pull request, #23357:
URL: https://github.com/apache/kafka/pull/23357
The async consumer, the share consumer and the streams consumer
busy-spin on both the network thread and the application thread while a
group heartbeat request is in flight and the heartbeat timer has
expired. This happens on the first heartbeat of every consumer instance,
and on any later heartbeat whose response takes longer than the
heartbeat interval. Both threads spin until the response or a failure
arrives.
### How it happens
`HeartbeatRequestState` resets the heartbeat timer when a heartbeat
request is generated and marked in flight (it may still be queued for an
unreachable host). While that request is in flight the manager cannot
send another one, so `poll()` and `maximumTimeToWait()` fall back to
`timeToNextHeartbeatMs()`. If the timer is expired at that point, it
returns `remainingBackoffMs()`, which is measured from the last response
and, with default settings, is already 0.
The timer is expired while a request is in flight in two cases:
- **First heartbeat.** `heartbeatIntervalMs` is initialised to 0 and
only known from the first
response, so `resetTimer()` leaves the timer expired immediately.
- **Slow response.** The interval is known (say 5 s) but the response
takes longer than that,
for example a coordinator in a long GC pause. The timer expires with
the request still in
flight.
The heartbeat managers use that value for both wait times:
- `poll()` returns `PollResult(0)`, so `ConsumerNetworkThread` calls
`client.poll(0)` in a
tight loop.
- `maximumTimeToWait()` returns `min(pollTimer.remainingMs() / 2, 0) =
0`, so the application
thread does not block in `poll()` and wakes the network thread on
every iteration.
The guards added by KAFKA-20253 and KAFKA-20970 check
`coordinator().isEmpty()`. They do not apply here because the
coordinator is already known.
### When it happens and why it matters
The heartbeat interval lives in the client, so the window is per
consumer instance: it opens when a new `KafkaConsumer` sends its first
heartbeat and closes at the first heartbeat response. Later rejoins of
the same instance (rebalance, fencing, session timeout) do not reopen it
because the interval is already known. A coordinator move on its own
does not open it either; an already-joined consumer goes through the
coordinator-unknown path and rejoins with a known interval.
What makes the window long is the coordinator not answering the first
heartbeat. Until it does, every attempt spins for the whole in-flight
time and only rests during the retry backoff between attempts:
```
heartbeat sent
|
+-- spin until the response, or until request.timeout.ms
|
+-- failure: coordinator unknown, retry backoff,
rediscovery, heartbeat resent
|
+-- spin again ... until the first response arrives
```
Ordered by impact:
1. **Consumer restarts while the coordinator is moving.** A rolling
broker upgrade moves the
coordinator role between brokers. Consumer pods often restart in the
same window (same
pipeline, autoscaling, liveness probes reacting to broker churn).
Every restarted consumer
is a new instance, and its first heartbeat goes to a coordinator that
accepts the
connection but does not answer, or to a host that is not reachable
yet. Both threads spin
until `request.timeout.ms` (30 s), or until the connection attempt
fails if the host is not
reachable; the failure marks the coordinator unknown, the normal
retry and rediscovery apply, and the next heartbeat
spins again until it is answered, for as long as the failover takes.
2. **Coordinator overloaded or in a long GC pause.** Any consumer whose
heartbeat response
takes longer than the broker-provided heartbeat interval (5 s by
default) spins for the rest of the pause,
up to `request.timeout.ms`. This does not need a restart;
already-joined consumers are
affected too.
3. **Coordinator not reachable from the client.** Wrong
`advertised.listeners` (common with
Docker and Kubernetes), or a firewall that only allows the bootstrap
broker. The consumer
never recovers and repeats the cycle above. Request and connection
timeouts are logged at
INFO/WARN, but nothing identifies the busy loop; the visible symptom
is high CPU and no
assignment, which is easy to misread as a load problem.
4. **Every normal start.** A few milliseconds of spin per instance on a
healthy cluster. Not
visible, but it shows the path is taken by every consumer.
Case 3 is a misconfiguration, but the client is designed to retry
forever because it cannot tell a transient failure from a permanent one.
While it retries it should wait quietly. Today it keeps two threads busy
per consumer, which starves co-located workloads and hides the real
cause.
### Fix
`HeartbeatRequestState.timeToNextHeartbeatMs()` returns the effective
initial retry backoff
(`min(retry.backoff.ms, retry.backoff.max.ms)`, 100 ms by default,
without jitter, floored at
1 ms) when the timer is expired and a request is in flight, instead of
the remaining backoff. In that state nothing
can be sent until the in-flight request completes, so the managers
re-check every backoff
interval. This is the in-flight wait only; after a failure the normal
jittered, exponential retry
and coordinator rediscovery apply. The change is complementary to
KAFKA-21010 (#23348): a
request timeout marks the coordinator unknown and moves the client from
this known-coordinator,
in-flight state to the unknown-coordinator state that #23348 (pending)
targets.
The change does not alter the heartbeat cadence: in the normal case the
timer is reset with the known interval at send time and is not expired
while the request is in flight, so this branch is never taken. It only
replaces the 0 in the two cases above.
One change covers the consumer, share and streams heartbeat managers,
and avoids adding a third special case next to the existing guards in
`maximumTimeToWait()`.
#### Why the retry backoff
Values considered for this state:
- **0 (trunk).** Busy loop.
- **`heartbeatIntervalMs`.** It is 0 in the first-heartbeat case, which
is how KAFKA-21010
happened.
- **`Long.MAX_VALUE` ("wait for I/O only").** `NetworkClient.poll()`
only detects a timed-out
request after `selector.poll()` returns, and `ConsumerNetworkThread`
caps that poll at 5 s, so a
heartbeat that hits `request.timeout.ms` would be noticed up to 5 s
late, delaying coordinator
rediscovery.
- **`retry.backoff.ms`.** Bounded: a request timeout is noticed within
one backoff. The cost is
one wakeup per backoff on both threads while a heartbeat is in flight,
which in the healthy
case is a few milliseconds per join. It is also the value KAFKA-21010
(#23348) uses for the
coordinator-unknown guard, so the heartbeat managers wait the same way
in both "cannot send"
and "already sent" states. `retry.backoff.ms` and
`retry.backoff.max.ms` both accept 0, so
the value is floored at 1 ms (`MIN_IN_FLIGHT_WAIT_MS`); the sibling
guards in #23348 do not
floor, which is worth aligning there.
The typed version of this decision (a manager saying "awaiting input"
instead of a number) is what KIP-1371 proposes; this change keeps the
current `long` contract.
### Testing
All eleven new regression-test invocations fail on trunk and pass with
the fix (verified on trunk plus the test commit alone, same failures on
every run). They use the same shapes as the tests added by KAFKA-20253
and KAFKA-21010.
-
`HeartbeatRequestStateTest.testTimeToNextHeartbeatMsWhileRequestInFlightDoesNotSpin`,
parameterised over a heartbeat interval of 0 (first heartbeat) and
5000 (response slower
than the interval), plus `...WithZeroRetryBackoffDoesNotSpin` for a
zero backoff. They isolate
the root cause on the state object and assert the exact value
(`retry.backoff.ms`, or the
1 ms floor as a literal), in the style of the KAFKA-20253 tests.
- `ConsumerHeartbeatRequestManagerTest`,
`ShareHeartbeatRequestManagerTest` and
`StreamsGroupHeartbeatRequestManagerTest.testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin`,
parameterised the same way. They send a heartbeat, leave it in flight
past the interval, and
assert that `poll()` sends nothing and returns `retry.backoff.ms` as
`timeUntilNextPollMs`,
and that `maximumTimeToWait()` returns the same.
-
`ConsumerHeartbeatRequestManagerTest.testMaximumTimeToWaitDoesNotSpinWhileHeartbeatInFlightOnRealNetworkClient`,
parameterised the same way, uses the same wiring as
`testMaximumTimeToWaitDoesNotSpinDuringRealBootstrapDnsResolution`
(real `NetworkClient` on
`MockSelector`, real `ConsumerMetadata` and `NetworkClientDelegate`)
with a known coordinator.
The heartbeat is really sent and never answered. The network poll is
driven with the manager's
own wait time, as `ConsumerNetworkThread.runOnce()` does, so the mock
clock only advances when
that wait is positive: the test bounds the number of polls while the
request is in flight
(about one per retry backoff) and fails with an iteration-count
message on trunk, where the
clock never moves. It then checks that the request times out at
`request.timeout.ms`, the
in-flight state is cleared, and a second heartbeat is sent.
Related: KAFKA-20253 (#22836), KAFKA-20970 (#23227), KAFKA-21010
(#23348). Those cover the coordinator-unknown cases; this change covers
the request-in-flight case, which they do not reach. The bug is present
since 3.7.0 (KAFKA-15278, KAFKA-15890; the send-time timer reset since
3.8.0, KAFKA-16528) and is not a 4.4 regression. Trunk is
4.5.0-SNAPSHOT; since 4.4 is still in the RC phase, including this in
the next 4.4 RC is proposed, plus cherry-picks to 4.3 and 4.2 as was
done for KAFKA-20253.
Generated-by: Claude Fable 5.1 Claude-Session:
https://claude.ai/code/session_015fJAXoLEZWgC25avikRtzg
--
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]
