lianetm commented on code in PR #23357:
URL: https://github.com/apache/kafka/pull/23357#discussion_r4037560252
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestState.java:
##########
@@ -70,6 +77,17 @@ public void resetTimer() {
public long timeToNextHeartbeatMs(final long currentTimeMs) {
if (heartbeatTimer.isExpired()) {
+ if (requestInFlight()) {
+ // The timer can be expired while a request is in flight both
for the first heartbeat (the
+ // interval is initialised to 0 and only learned from the
first response) and for any later
+ // heartbeat whose response takes longer than the interval. No
heartbeat can be sent until
+ // the in-flight one completes. The remaining backoff is
measured from the last response and,
+ // with default settings, is already 0 here, which would
busy-spin both the application and
+ // network threads. Wait the initial retry backoff (floored at
1 ms) instead of waiting forever: the network thread still has to notice a
request timeout promptly
+ // (NetworkClient only checks timed-out requests after
selector.poll returns, and
+ // ConsumerNetworkThread caps the poll at 5s), so it must come
back and re-check.
+ return Math.max(MIN_IN_FLIGHT_WAIT_MS,
exponentialBackoff.initialInterval());
Review Comment:
let's use the accessor `retryBackoffMs()` ?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -446,6 +451,348 @@ consumerMetadata, mock(BackgroundEventHandler.class),
false, mock(AsyncConsumerM
}
}
+ /**
+ * A heartbeat request is in flight and the heartbeat timer is already
expired. That happens both
+ * while the very first heartbeat is in flight, when the interval is still
unknown (it is initialised
+ * to 0 and only learned from the first heartbeat response), and later on,
when a response takes
+ * longer than the interval. In that window no heartbeat can be sent until
the in-flight one
+ * completes, so both {@link
NetworkClientDelegate.PollResult#timeUntilNextPollMs} and
+ * {@link AbstractHeartbeatRequestManager#maximumTimeToWait(long)} must
return a positive delay;
+ * returning 0 busy-spins the consumer network thread and the application
thread until the in-flight
+ * request completes, which can be as long as request.timeout.ms when the
coordinator is unreachable.
+ */
+ @ParameterizedTest
+ @ValueSource(longs = {0, 5000})
+ public void testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin(final
long heartbeatIntervalMs) {
+ createHeartbeatRequestStateWithHeartbeatInterval(heartbeatIntervalMs);
+ // The member keeps joining for both intervals, so the heartbeat below
is sent without waiting for
+ // the interval and the total simulated time stays under
max.poll.interval.ms.
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(true);
+ if (heartbeatIntervalMs > 0) {
+ // A non-zero interval is only known after a heartbeat response,
which also arms the backoff.
+ heartbeatRequestState.onSuccessfulAttempt(time.milliseconds());
+ }
Review Comment:
why only for hb interval > 0? `onSuccessfulAttempt` happens when we send ,
not when we receive the response (so I expect it's right to call it no matter
the HB interval we have here?)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestState.java:
##########
@@ -28,6 +28,13 @@
*/
public class HeartbeatRequestState extends RequestState {
+ /**
+ * Lower bound for the wait returned while a heartbeat is in flight.
retry.backoff.ms and
+ * retry.backoff.max.ms both accept 0, and callers use this value as a
poll timeout, so it must
+ * never be 0.
+ */
+ private static final long MIN_IN_FLIGHT_WAIT_MS = 1L;
Review Comment:
I get the point with this, but it introduces an inconsistency, so I would
say we either fix consistently (all req managers) or don't (we haven't used any
floor in any of the other paths fixed IIRC)
`retry.backoff.ms=0` would be a very extreme/unexpected config setting I
expect, so I wouldn't rush in any extra logic/complexity in the managers to
guard against it, wdyt? We can always file something to review if it's worth
any change but address it consistently
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -446,6 +451,348 @@ consumerMetadata, mock(BackgroundEventHandler.class),
false, mock(AsyncConsumerM
}
}
+ /**
+ * A heartbeat request is in flight and the heartbeat timer is already
expired. That happens both
+ * while the very first heartbeat is in flight, when the interval is still
unknown (it is initialised
+ * to 0 and only learned from the first heartbeat response), and later on,
when a response takes
+ * longer than the interval. In that window no heartbeat can be sent until
the in-flight one
+ * completes, so both {@link
NetworkClientDelegate.PollResult#timeUntilNextPollMs} and
+ * {@link AbstractHeartbeatRequestManager#maximumTimeToWait(long)} must
return a positive delay;
+ * returning 0 busy-spins the consumer network thread and the application
thread until the in-flight
+ * request completes, which can be as long as request.timeout.ms when the
coordinator is unreachable.
+ */
+ @ParameterizedTest
+ @ValueSource(longs = {0, 5000})
+ public void testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin(final
long heartbeatIntervalMs) {
+ createHeartbeatRequestStateWithHeartbeatInterval(heartbeatIntervalMs);
+ // The member keeps joining for both intervals, so the heartbeat below
is sent without waiting for
+ // the interval and the total simulated time stays under
max.poll.interval.ms.
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(true);
+ if (heartbeatIntervalMs > 0) {
+ // A non-zero interval is only known after a heartbeat response,
which also arms the backoff.
Review Comment:
this comment is a bit confusing too, as `onSuccessfulAttempt` happens when
we send (maybe remove if my other comment makes sense and we end up calling
this regardless of the interval
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -446,6 +451,348 @@ consumerMetadata, mock(BackgroundEventHandler.class),
false, mock(AsyncConsumerM
}
}
+ /**
+ * A heartbeat request is in flight and the heartbeat timer is already
expired. That happens both
+ * while the very first heartbeat is in flight, when the interval is still
unknown (it is initialised
+ * to 0 and only learned from the first heartbeat response), and later on,
when a response takes
+ * longer than the interval. In that window no heartbeat can be sent until
the in-flight one
+ * completes, so both {@link
NetworkClientDelegate.PollResult#timeUntilNextPollMs} and
+ * {@link AbstractHeartbeatRequestManager#maximumTimeToWait(long)} must
return a positive delay;
+ * returning 0 busy-spins the consumer network thread and the application
thread until the in-flight
+ * request completes, which can be as long as request.timeout.ms when the
coordinator is unreachable.
+ */
+ @ParameterizedTest
+ @ValueSource(longs = {0, 5000})
+ public void testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin(final
long heartbeatIntervalMs) {
+ createHeartbeatRequestStateWithHeartbeatInterval(heartbeatIntervalMs);
+ // The member keeps joining for both intervals, so the heartbeat below
is sent without waiting for
+ // the interval and the total simulated time stays under
max.poll.interval.ms.
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(true);
+ if (heartbeatIntervalMs > 0) {
+ // A non-zero interval is only known after a heartbeat response,
which also arms the backoff.
+ heartbeatRequestState.onSuccessfulAttempt(time.milliseconds());
+ }
+
+ NetworkClientDelegate.PollResult firstResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, firstResult.unsentRequests.size(),
+ "A heartbeat should be sent as soon as the coordinator is known");
+
+ // Deliberately do not complete the request, so it stays in flight
while the heartbeat timer expires.
+ time.sleep(heartbeatIntervalMs + 1);
+
+ NetworkClientDelegate.PollResult secondResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(0, secondResult.unsentRequests.size(),
+ "No heartbeat should be sent while another one is in flight");
+ assertTrue(secondResult.timeUntilNextPollMs > 0,
+ "timeUntilNextPollMs must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got "
+ + secondResult.timeUntilNextPollMs);
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS,
secondResult.timeUntilNextPollMs);
+
+ long result =
heartbeatRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got " + result);
+ // maximumTimeToWait is min(pollTimer.remainingMs() / 2, retry
backoff), and half of the remaining
+ // max.poll.interval.ms is still larger than the backoff at this point.
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS, result);
+ }
+
+ /**
+ * The "response slower than the interval" way of reaching the same
window, driven end to end through the
+ * manager instead of by priming the request state directly. The member
joins, learns its heartbeat interval
+ * from a real successful heartbeat response, becomes STABLE, and then
sends its steady-state heartbeat when
+ * the interval elapses. That response never arrives, so the heartbeat
timer expires again while the request
+ * is still in flight. This complements
+ * {@link #testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin(long)},
which constructs the request state
+ * with a known interval, by proving that the interval learned through
+ * {@code onResponse -> updateHeartbeatIntervalMs} lands the manager in
exactly the same state: no heartbeat
+ * can be sent, and both {@link
NetworkClientDelegate.PollResult#timeUntilNextPollMs} and
+ * {@link AbstractHeartbeatRequestManager#maximumTimeToWait(long)} must
return a positive delay rather than
+ * busy-spinning the application and network threads.
+ */
+ @Test
+ public void
testMaximumTimeToWaitWhenResponseIsSlowerThanIntervalDoesNotSpin() {
+ // The interval is unknown until the first heartbeat response, exactly
as on a freshly created consumer.
+ createHeartbeatRequestStateWithZeroHeartbeatInterval();
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(true);
+
+ NetworkClientDelegate.PollResult joinResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, joinResult.unsentRequests.size(),
+ "A heartbeat should be sent as soon as the coordinator is known");
+
+ // A real successful response teaches the manager the interval and
clears the in-flight flag.
+ joinResult.unsentRequests.get(0).handler().onComplete(
+ createHeartbeatResponse(joinResult.unsentRequests.get(0),
Errors.NONE, DEFAULT_HEARTBEAT_INTERVAL_MS));
+ assertEquals(DEFAULT_HEARTBEAT_INTERVAL_MS,
heartbeatRequestState.heartbeatIntervalMs(),
+ "The heartbeat interval should have been learned from the
heartbeat response");
+
+ // The membership manager is a mock, so onHeartbeatSuccess does not
move it; stub the joined member state.
+ when(membershipManager.state()).thenReturn(MemberState.STABLE);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(false);
+ when(membershipManager.shouldSkipHeartbeat()).thenReturn(false);
+
+ // The interval elapses, so the steady-state heartbeat is sent.
Deliberately leave it in flight.
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
+ NetworkClientDelegate.PollResult heartbeatResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, heartbeatResult.unsentRequests.size(),
+ "A heartbeat should be sent once the heartbeat interval has
expired");
+
+ // The response is slower than the interval, so the heartbeat timer
expires again while it is in flight.
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS + 1);
+
+ NetworkClientDelegate.PollResult inFlightResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(0, inFlightResult.unsentRequests.size(),
+ "No heartbeat should be sent while another one is in flight");
+ assertTrue(inFlightResult.timeUntilNextPollMs > 0,
+ "timeUntilNextPollMs must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got "
+ + inFlightResult.timeUntilNextPollMs);
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS,
inFlightResult.timeUntilNextPollMs);
+
+ long result =
heartbeatRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got " + result);
+ // maximumTimeToWait is min(pollTimer.remainingMs() / 2, retry
backoff), and half of the remaining
+ // max.poll.interval.ms is still larger than the backoff at this point.
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS, result);
+ }
+
+ /**
+ * The same busy-spin, driven through a real {@link NetworkClient} on a
{@link MockSelector} and a real
+ * {@link NetworkClientDelegate}, so that the heartbeat really is in
flight rather than only marked as such.
Review Comment:
uhm do we really need this test? We're unit testing the req manager and we
already tested the in-flight case above, but seems this test is to cover the
same case but going beyond to the other component (network client)? If so I
would say we shouldn't get there and remove this?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -446,6 +451,348 @@ consumerMetadata, mock(BackgroundEventHandler.class),
false, mock(AsyncConsumerM
}
}
+ /**
+ * A heartbeat request is in flight and the heartbeat timer is already
expired. That happens both
+ * while the very first heartbeat is in flight, when the interval is still
unknown (it is initialised
+ * to 0 and only learned from the first heartbeat response), and later on,
when a response takes
+ * longer than the interval. In that window no heartbeat can be sent until
the in-flight one
+ * completes, so both {@link
NetworkClientDelegate.PollResult#timeUntilNextPollMs} and
+ * {@link AbstractHeartbeatRequestManager#maximumTimeToWait(long)} must
return a positive delay;
+ * returning 0 busy-spins the consumer network thread and the application
thread until the in-flight
+ * request completes, which can be as long as request.timeout.ms when the
coordinator is unreachable.
+ */
+ @ParameterizedTest
+ @ValueSource(longs = {0, 5000})
+ public void testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin(final
long heartbeatIntervalMs) {
+ createHeartbeatRequestStateWithHeartbeatInterval(heartbeatIntervalMs);
+ // The member keeps joining for both intervals, so the heartbeat below
is sent without waiting for
+ // the interval and the total simulated time stays under
max.poll.interval.ms.
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(true);
+ if (heartbeatIntervalMs > 0) {
+ // A non-zero interval is only known after a heartbeat response,
which also arms the backoff.
+ heartbeatRequestState.onSuccessfulAttempt(time.milliseconds());
+ }
+
+ NetworkClientDelegate.PollResult firstResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, firstResult.unsentRequests.size(),
+ "A heartbeat should be sent as soon as the coordinator is known");
+
+ // Deliberately do not complete the request, so it stays in flight
while the heartbeat timer expires.
+ time.sleep(heartbeatIntervalMs + 1);
+
+ NetworkClientDelegate.PollResult secondResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(0, secondResult.unsentRequests.size(),
+ "No heartbeat should be sent while another one is in flight");
+ assertTrue(secondResult.timeUntilNextPollMs > 0,
+ "timeUntilNextPollMs must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got "
+ + secondResult.timeUntilNextPollMs);
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS,
secondResult.timeUntilNextPollMs);
+
+ long result =
heartbeatRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got " + result);
+ // maximumTimeToWait is min(pollTimer.remainingMs() / 2, retry
backoff), and half of the remaining
+ // max.poll.interval.ms is still larger than the backoff at this point.
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS, result);
+ }
+
+ /**
+ * The "response slower than the interval" way of reaching the same
window, driven end to end through the
+ * manager instead of by priming the request state directly. The member
joins, learns its heartbeat interval
+ * from a real successful heartbeat response, becomes STABLE, and then
sends its steady-state heartbeat when
+ * the interval elapses. That response never arrives, so the heartbeat
timer expires again while the request
+ * is still in flight. This complements
+ * {@link #testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin(long)},
which constructs the request state
+ * with a known interval, by proving that the interval learned through
+ * {@code onResponse -> updateHeartbeatIntervalMs} lands the manager in
exactly the same state: no heartbeat
+ * can be sent, and both {@link
NetworkClientDelegate.PollResult#timeUntilNextPollMs} and
+ * {@link AbstractHeartbeatRequestManager#maximumTimeToWait(long)} must
return a positive delay rather than
+ * busy-spinning the application and network threads.
+ */
+ @Test
+ public void
testMaximumTimeToWaitWhenResponseIsSlowerThanIntervalDoesNotSpin() {
+ // The interval is unknown until the first heartbeat response, exactly
as on a freshly created consumer.
+ createHeartbeatRequestStateWithZeroHeartbeatInterval();
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(true);
+
+ NetworkClientDelegate.PollResult joinResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, joinResult.unsentRequests.size(),
+ "A heartbeat should be sent as soon as the coordinator is known");
+
+ // A real successful response teaches the manager the interval and
clears the in-flight flag.
+ joinResult.unsentRequests.get(0).handler().onComplete(
+ createHeartbeatResponse(joinResult.unsentRequests.get(0),
Errors.NONE, DEFAULT_HEARTBEAT_INTERVAL_MS));
+ assertEquals(DEFAULT_HEARTBEAT_INTERVAL_MS,
heartbeatRequestState.heartbeatIntervalMs(),
+ "The heartbeat interval should have been learned from the
heartbeat response");
+
+ // The membership manager is a mock, so onHeartbeatSuccess does not
move it; stub the joined member state.
+ when(membershipManager.state()).thenReturn(MemberState.STABLE);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(false);
+ when(membershipManager.shouldSkipHeartbeat()).thenReturn(false);
+
+ // The interval elapses, so the steady-state heartbeat is sent.
Deliberately leave it in flight.
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
+ NetworkClientDelegate.PollResult heartbeatResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, heartbeatResult.unsentRequests.size(),
+ "A heartbeat should be sent once the heartbeat interval has
expired");
+
+ // The response is slower than the interval, so the heartbeat timer
expires again while it is in flight.
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS + 1);
+
+ NetworkClientDelegate.PollResult inFlightResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(0, inFlightResult.unsentRequests.size(),
+ "No heartbeat should be sent while another one is in flight");
+ assertTrue(inFlightResult.timeUntilNextPollMs > 0,
+ "timeUntilNextPollMs must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got "
+ + inFlightResult.timeUntilNextPollMs);
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS,
inFlightResult.timeUntilNextPollMs);
+
+ long result =
heartbeatRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got " + result);
+ // maximumTimeToWait is min(pollTimer.remainingMs() / 2, retry
backoff), and half of the remaining
+ // max.poll.interval.ms is still larger than the backoff at this point.
Review Comment:
if my [comment
above](https://github.com/apache/kafka/pull/23357/changes#r4037628360) mnakes
sense we need to update here too.
Actually maybe helpful to encapsulate the 2 assertion blocks (the 2 tests
generate different scenarios but have the same expectations timeToNextPoll, ln
552-557 here and maximumTimeToWait, ln 544-550 right?)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestState.java:
##########
@@ -70,6 +77,17 @@ public void resetTimer() {
public long timeToNextHeartbeatMs(final long currentTimeMs) {
if (heartbeatTimer.isExpired()) {
+ if (requestInFlight()) {
+ // The timer can be expired while a request is in flight both
for the first heartbeat (the
+ // interval is initialised to 0 and only learned from the
first response) and for any later
+ // heartbeat whose response takes longer than the interval. No
heartbeat can be sent until
+ // the in-flight one completes. The remaining backoff is
measured from the last response and,
+ // with default settings, is already 0 here, which would
busy-spin both the application and
+ // network threads. Wait the initial retry backoff (floored at
1 ms) instead of waiting forever: the network thread still has to notice a
request timeout promptly
+ // (NetworkClient only checks timed-out requests after
selector.poll returns, and
+ // ConsumerNetworkThread caps the poll at 5s), so it must come
back and re-check.
Review Comment:
could we simplify this long comment?
Probably ok to just highlight the "why" (timer can be expired while
in-flight, and remaining backoff can be already 0, measured from the last
response), so we return backoff to check again after a bit and avoid busy-spin)
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -446,6 +451,348 @@ consumerMetadata, mock(BackgroundEventHandler.class),
false, mock(AsyncConsumerM
}
}
+ /**
+ * A heartbeat request is in flight and the heartbeat timer is already
expired. That happens both
+ * while the very first heartbeat is in flight, when the interval is still
unknown (it is initialised
+ * to 0 and only learned from the first heartbeat response), and later on,
when a response takes
+ * longer than the interval. In that window no heartbeat can be sent until
the in-flight one
+ * completes, so both {@link
NetworkClientDelegate.PollResult#timeUntilNextPollMs} and
+ * {@link AbstractHeartbeatRequestManager#maximumTimeToWait(long)} must
return a positive delay;
+ * returning 0 busy-spins the consumer network thread and the application
thread until the in-flight
+ * request completes, which can be as long as request.timeout.ms when the
coordinator is unreachable.
+ */
+ @ParameterizedTest
+ @ValueSource(longs = {0, 5000})
+ public void testMaximumTimeToWaitWhileHeartbeatInFlightDoesNotSpin(final
long heartbeatIntervalMs) {
+ createHeartbeatRequestStateWithHeartbeatInterval(heartbeatIntervalMs);
+ // The member keeps joining for both intervals, so the heartbeat below
is sent without waiting for
+ // the interval and the total simulated time stays under
max.poll.interval.ms.
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ when(membershipManager.shouldHeartbeatNow()).thenReturn(true);
+ if (heartbeatIntervalMs > 0) {
+ // A non-zero interval is only known after a heartbeat response,
which also arms the backoff.
+ heartbeatRequestState.onSuccessfulAttempt(time.milliseconds());
+ }
+
+ NetworkClientDelegate.PollResult firstResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, firstResult.unsentRequests.size(),
+ "A heartbeat should be sent as soon as the coordinator is known");
+
+ // Deliberately do not complete the request, so it stays in flight
while the heartbeat timer expires.
+ time.sleep(heartbeatIntervalMs + 1);
+
+ NetworkClientDelegate.PollResult secondResult =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(0, secondResult.unsentRequests.size(),
+ "No heartbeat should be sent while another one is in flight");
+ assertTrue(secondResult.timeUntilNextPollMs > 0,
+ "timeUntilNextPollMs must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got "
+ + secondResult.timeUntilNextPollMs);
+ assertEquals(DEFAULT_RETRY_BACKOFF_MS,
secondResult.timeUntilNextPollMs);
+
+ long result =
heartbeatRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 while a heartbeat is in flight to
avoid a busy-spin; got " + result);
+ // maximumTimeToWait is min(pollTimer.remainingMs() / 2, retry
backoff), and half of the remaining
+ // max.poll.interval.ms is still larger than the backoff at this point.
Review Comment:
is this comment needed or could we remove? seems like it can easily become
stale if we make any changes to the implementation.
--
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]
