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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
