lianetm commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3961504224
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -245,27 +245,36 @@ public PollResult pollOnClose(long currentTimeMs) {
* <p>Similarly, we may have to unblock the application thread to send a
{@link AsyncPollEvent} to make sure
* our poll timer will not expire while we are polling.
*
- * <p>When the member is {@link MemberState#UNSUBSCRIBED} (for example,
with manual assignment),
- * this returns {@code Long.MAX_VALUE} to indicate there is no next
heartbeat to wait for,
- * allowing the application thread to block for the full user-specified
poll timeout rather than
- * spinning in a busy loop.
+ * <p>When the member is {@link MemberState#UNSUBSCRIBED} or in the
terminal {@link MemberState#FATAL} state,
+ * this returns {@code Long.MAX_VALUE} to indicate there is no next
heartbeat to wait for, allowing the application
+ * thread to block for the full user-specified poll timeout rather than
spinning in a busy loop.
*/
@Override
public long maximumTimeToWait(long currentTimeMs) {
pollTimer.update(currentTimeMs);
- if (membershipManager().state() == MemberState.UNSUBSCRIBED) {
+ MemberState state = membershipManager().state();
+ // No heartbeat can be sent in these states: UNSUBSCRIBED has nothing
to heartbeat for,
+ // and FATAL is terminal. The fatal error has already been propagated
to the
+ // application thread, so there is no need to wake it before its poll
timeout expires.
+ if (state == MemberState.UNSUBSCRIBED || state == MemberState.FATAL) {
return Long.MAX_VALUE;
}
if (pollTimer.isExpired()) {
return 0L;
}
Review Comment:
is it worth a comment that here we need to unblock the app thread so that
the stale consumer runs callbacks if needed and allows a call to poll again to
rejoin?
##########
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:
aligned with the FATAL case, thanks for the update!
About STALE:
> stale member always has an expired poll timer
well technically they don't move together, right? and actually you have a
comment further down with `"STALE with the poll timer already reset"`).
I expect we could indeed be STALE + timer already reset (a next poll after
STALE will first send the `AsyncPollEvent` that resets the timer, but the
transition out of stale won't happen until the callbacks are processed,
separately in the app thread with `processBackgroundEvents` (even bigger window
if there are commit callbacks enqueued). If this expectation is correct, then
the STALE case is indeed reachable here, with timer not expired, and we would
be returning the `retryBackoff`, correct? Different than the 0 returned if
pollTimer is expired. For poll timer expired, STALE and FENCED, don't we want
the same you said : "desired behavior because it hands control back to the
application thread" (to run callbacks / trigger rejoin)
I don't want to derail this PR, so I would be ok with shipping it with the
fix it has:
- return retryBackoff for empty coord (this was the core issue)
- return retryBackoff for STALE/FENCED (which is better than the interval 0
or actual interval it had before this PR)
- as follow-up, review the STALE/FENCE case to see if we can improve
further? or is there an clear busy loop we see if we return 0 for STALE/FENCED
(to go back to the app thread, callbacks, then JOINING)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -245,27 +245,36 @@ public PollResult pollOnClose(long currentTimeMs) {
* <p>Similarly, we may have to unblock the application thread to send a
{@link AsyncPollEvent} to make sure
* our poll timer will not expire while we are polling.
*
- * <p>When the member is {@link MemberState#UNSUBSCRIBED} (for example,
with manual assignment),
- * this returns {@code Long.MAX_VALUE} to indicate there is no next
heartbeat to wait for,
- * allowing the application thread to block for the full user-specified
poll timeout rather than
- * spinning in a busy loop.
+ * <p>When the member is {@link MemberState#UNSUBSCRIBED} or in the
terminal {@link MemberState#FATAL} state,
+ * this returns {@code Long.MAX_VALUE} to indicate there is no next
heartbeat to wait for, allowing the application
+ * thread to block for the full user-specified poll timeout rather than
spinning in a busy loop.
*/
@Override
public long maximumTimeToWait(long currentTimeMs) {
pollTimer.update(currentTimeMs);
- if (membershipManager().state() == MemberState.UNSUBSCRIBED) {
+ MemberState state = membershipManager().state();
+ // No heartbeat can be sent in these states: UNSUBSCRIBED has nothing
to heartbeat for,
+ // and FATAL is terminal. The fatal error has already been propagated
to the
+ // application thread, so there is no need to wake it before its poll
timeout expires.
+ if (state == MemberState.UNSUBSCRIBED || state == MemberState.FATAL) {
return Long.MAX_VALUE;
}
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. Otherwise, the timer-based
branches below may return 0
+ // (the heartbeat timer remains permanently expired), causing both the
application and network
+ // threads to busy-spin.
Review Comment:
is all this comment about "why not using HB internval" redundant now with
what's in ln 274/275 ("Return retryBackoffMs rather than the heartbeat
interval.....").
--
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]
