m1a2st commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3944351468
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -259,13 +259,16 @@ public long maximumTimeToWait(long currentTimeMs) {
if (pollTimer.isExpired()) {
return 0L;
}
- // KAFKA-20253: mirror the guard in poll(). A heartbeat is only sent
when the coordinator is known
- // and the member is in a state that heartbeats. When the coordinator
is unavailable (e.g. after a
- // re-authentication failure) or the member should skip heartbeats
(FATAL/FENCED/STALE/UNSUBSCRIBED),
- // poll() returns EMPTY, so falling through to the timer-based
branches below would return 0 (the
- // heartbeat timer is left permanently expired) and busy-spin both the
application and network threads.
+ // Mirror the guard in poll(). A heartbeat is only sent when the
coordinator is known and the
+ // member is in a state that heartbeats. When the coordinator is
unavailable (e.g. after a
+ // re-authentication failure, or while bootstrap DNS resolution is
still in progress) or the
+ // member should skip heartbeats (FATAL/FENCED/STALE/UNSUBSCRIBED),
poll() returns EMPTY, so
+ // falling through to the timer-based branches below would return 0
(the heartbeat timer is left
+ // permanently expired) and busy-spin both the application and network
threads. Wait a retry
+ // backoff rather than the heartbeat interval, because the interval is
zero until the first
+ // heartbeat response is received, which would also busy-spin.
if (coordinatorRequestManager.coordinator().isEmpty() ||
membershipManager().shouldSkipHeartbeat()) {
- return heartbeatRequestState.heartbeatIntervalMs();
+ return heartbeatRequestState.retryBackoffMs();
Review Comment:
Thanks for the point out!
`FATAL` is a terminal state and the error has already been queued for the
application thread, so I moved it into the early-return path with
`UNSUBSCRIBED`, returning.
`STALE` is effectively unreachable here. A stale member always has an
expired poll timer, so the expired-timer check above returns `0` first, which
is the desired behavior because it hands control back to the application thread
and allows the next poll to trigger the rejoin. The only exception is the brief
window after the timer has been reset but before the lost-partitions callback
completes. In that case, the member is effectively in the same situation as
`FENCED`, so the retry backoff remains appropriate.
For `FENCED`, the member cannot rejoin until the lost-partitions callback
runs on the application thread, and background events do not wake a blocked
application thread. This return value therefore provides the only bound on that
wait; returning `Long.MAX_VALUE` could delay the rejoin until the full poll
timeout expires.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -259,13 +259,16 @@ public long maximumTimeToWait(long currentTimeMs) {
if (pollTimer.isExpired()) {
return 0L;
}
- // KAFKA-20253: mirror the guard in poll(). A heartbeat is only sent
when the coordinator is known
- // and the member is in a state that heartbeats. When the coordinator
is unavailable (e.g. after a
- // re-authentication failure) or the member should skip heartbeats
(FATAL/FENCED/STALE/UNSUBSCRIBED),
- // poll() returns EMPTY, so falling through to the timer-based
branches below would return 0 (the
- // heartbeat timer is left permanently expired) and busy-spin both the
application and network threads.
+ // Mirror the guard in poll(). A heartbeat is only sent when the
coordinator is known and the
+ // member is in a state that heartbeats. When the coordinator is
unavailable (e.g. after a
+ // re-authentication failure, or while bootstrap DNS resolution is
still in progress) or the
+ // member should skip heartbeats (FATAL/FENCED/STALE/UNSUBSCRIBED),
poll() returns EMPTY, so
+ // falling through to the timer-based branches below would return 0
(the heartbeat timer is left
+ // permanently expired) and busy-spin both the application and network
threads. Wait a retry
+ // backoff rather than the heartbeat interval, because the interval is
zero until the first
+ // heartbeat response is received, which would also busy-spin.
if (coordinatorRequestManager.coordinator().isEmpty() ||
membershipManager().shouldSkipHeartbeat()) {
- return heartbeatRequestState.heartbeatIntervalMs();
+ return heartbeatRequestState.retryBackoffMs();
Review Comment:
Thanks for the point out!
`FATAL` is a terminal state and the error has already been queued for the
application thread, so I moved it into the early-return path with
`UNSUBSCRIBED`.
`STALE` is effectively unreachable here. A stale member always has an
expired poll timer, so the expired-timer check above returns `0` first, which
is the desired behavior because it hands control back to the application thread
and allows the next poll to trigger the rejoin. The only exception is the brief
window after the timer has been reset but before the lost-partitions callback
completes. In that case, the member is effectively in the same situation as
`FENCED`, so the retry backoff remains appropriate.
For `FENCED`, the member cannot rejoin until the lost-partitions callback
runs on the application thread, and background events do not wake a blocked
application thread. This return value therefore provides the only bound on that
wait; returning `Long.MAX_VALUE` could delay the rejoin until the full poll
timeout expires.
--
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]
