You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2026/09/03 16:14:24 UTC
[PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
m1a2st opened a new pull request, #23348:
URL: https://github.com/apache/kafka/pull/23348
Follow-up to (#23227).
Before bootstrap DNS resolution completes, `coordinatorRequestManager.coordinator()` is empty, causing the coordinator-unknown path in `maximumTimeToWait()` to return `heartbeatRequestState.heartbeatIntervalMs()`. However, the heartbeat interval is initialized to 0 and is not updated until the first heartbeat response is received. As a result, a member in the JOINING state reports `maximumTimeToWait() == 0`.
Since the application thread's poll timeout is bounded by this value, both the application and network threads can busy-spin until the coordinator is discovered.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3971002995
##########
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:
Just for the record, ok with keeping the STALE/FENCE case as it is. My concern of the edge case where the callbacks land after the call to processBackgroundEvents but before the block on poll would simply need to wait for that block so that the next run triggers them (one poll cycle), which is the pattern/contract we have anyways around client reconciliations.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareHeartbeatRequestManagerTest.java:
##########
@@ -164,15 +164,46 @@ private void createHeartbeatStateAndRequestManager() {
@Test
public void testMaximumTimeToWaitWhenHeartbeatShouldBeSkippedDoesNotSpin() {
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(new Node(1, "localhost", 9999)));
- when(membershipManager.state()).thenReturn(MemberState.FATAL);
+ when(membershipManager.state()).thenReturn(MemberState.FENCED);
when(membershipManager.shouldSkipHeartbeat()).thenReturn(true);
when(heartbeatRequestState.timeToNextHeartbeatMs(anyLong())).thenReturn(0L);
long result = heartbeatRequestManager.maximumTimeToWait(time.milliseconds());
assertTrue(result > 0,
"maximumTimeToWait must be > 0 while heartbeats are skipped to avoid a busy-spin; got " + result);
- assertEquals(DEFAULT_HEARTBEAT_INTERVAL_MS, result);
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS, result);
+ }
+
+ @Test
+ public void testMaximumTimeToWaitWhenFatalReturnsMaxValue() {
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(new Node(1, "localhost", 9999)));
+ when(membershipManager.state()).thenReturn(MemberState.FATAL);
+
+ assertEquals(Long.MAX_VALUE, heartbeatRequestManager.maximumTimeToWait(time.milliseconds()),
+ "maximumTimeToWait should return Long.MAX_VALUE in the terminal FATAL state");
+ }
+
+ /**
+ * While bootstrap DNS resolution is still in progress the coordinator is unknown,
+ * and a member that wants to join has a zero heartbeat interval, since the interval is only
+ * learned from the first heartbeat response. maximumTimeToWait() must wait a retry backoff
+ * rather than the (zero) heartbeat interval; returning 0 busy-spins the application and
+ * network threads.
+ */
+ @Test
+ public void testMaximumTimeToWaitWhenJoiningAndCoordinatorUnknownDoesNotSpin() {
Review Comment:
the logic for maximumTimeToWait lives in the AbstractHBReqMgr, so I would expect we may be able to not repeat the tests for consumer and share? Filed https://issues.apache.org/jira/browse/KAFKA-21059 to follow-up separately in case we can clean up.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3963794645
##########
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:
Sure, added a comment for it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3937740607
##########
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:
Thinking about the case of shouldSkipHB.
That's the case of member:
- UNSUBCRIBED (already handled above btw, and by returning max_value, which makes sense, we're not waiting for any action on the HBMgr, so can let the app thread block for the remaining poll timeout).
- FATAL/STALE/FENCED -> these would now return retryBackoff, would that be correct?
- Fatal should probably be just as unsubscribed (max value), no need to have the app thread responsive to the req managers any sooner than the poll timeout (it's a final/unrecoverable state anyways)
- STALE/FENCED -> in those we just need action from the app thread really (poll to rejoin after stale, and run callbacks if fenced). Do we need to return a maxTime here to lower the app thread remaining poll timeout?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
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`.
`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.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #23348:
URL: https://github.com/apache/kafka/pull/23348
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
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 `Long.MAX_VALUE`.
`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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #23348:
URL: https://github.com/apache/kafka/pull/23348#issuecomment-5605820788
CHerry-picked to 4.4 https://github.com/apache/kafka/commit/d12e95da90259dd063cd736d9a9d579c54bf182e
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3974438349
##########
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:
Sorry for the late reply, I traced through this last night.
I don't have a better improvement in mind right now. For now, I'll open a Jira ticket to track this issue so we can discuss and investigate it separately.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3963788645
##########
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:
I have updated the comment
##########
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:
I have updated the comment.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3975192527
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareHeartbeatRequestManagerTest.java:
##########
@@ -164,15 +164,46 @@ private void createHeartbeatStateAndRequestManager() {
@Test
public void testMaximumTimeToWaitWhenHeartbeatShouldBeSkippedDoesNotSpin() {
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(new Node(1, "localhost", 9999)));
- when(membershipManager.state()).thenReturn(MemberState.FATAL);
+ when(membershipManager.state()).thenReturn(MemberState.FENCED);
when(membershipManager.shouldSkipHeartbeat()).thenReturn(true);
when(heartbeatRequestState.timeToNextHeartbeatMs(anyLong())).thenReturn(0L);
long result = heartbeatRequestManager.maximumTimeToWait(time.milliseconds());
assertTrue(result > 0,
"maximumTimeToWait must be > 0 while heartbeats are skipped to avoid a busy-spin; got " + result);
- assertEquals(DEFAULT_HEARTBEAT_INTERVAL_MS, result);
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS, result);
+ }
+
+ @Test
+ public void testMaximumTimeToWaitWhenFatalReturnsMaxValue() {
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(new Node(1, "localhost", 9999)));
+ when(membershipManager.state()).thenReturn(MemberState.FATAL);
+
+ assertEquals(Long.MAX_VALUE, heartbeatRequestManager.maximumTimeToWait(time.milliseconds()),
+ "maximumTimeToWait should return Long.MAX_VALUE in the terminal FATAL state");
+ }
+
+ /**
+ * While bootstrap DNS resolution is still in progress the coordinator is unknown,
+ * and a member that wants to join has a zero heartbeat interval, since the interval is only
+ * learned from the first heartbeat response. maximumTimeToWait() must wait a retry backoff
+ * rather than the (zero) heartbeat interval; returning 0 busy-spins the application and
+ * network threads.
+ */
+ @Test
+ public void testMaximumTimeToWaitWhenJoiningAndCoordinatorUnknownDoesNotSpin() {
Review Comment:
Thanks for bringing this up.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21010 Another busy loop happens if the member is in JOINING state and bootstrap DNS resolution is not finished [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23348:
URL: https://github.com/apache/kafka/pull/23348#discussion_r3978938912
##########
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:
FYI: https://issues.apache.org/jira/browse/KAFKA-21063
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org