lianetm commented on code in PR #16200:
URL: https://github.com/apache/kafka/pull/16200#discussion_r1683354345
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -659,95 +665,50 @@ public void
testPollTimerExpirationShouldNotMarkMemberStaleIfMemberAlreadyLeavin
@Test
public void testisExpiredByUsedForLogging() {
- Timer pollTimer = spy(time.timer(DEFAULT_MAX_POLL_INTERVAL_MS));
- heartbeatRequestManager = new HeartbeatRequestManager(new
LogContext(), pollTimer, config(),
- coordinatorRequestManager, membershipManager, heartbeatState,
heartbeatRequestState,
- backgroundEventHandler, metrics);
when(membershipManager.shouldSkipHeartbeat()).thenReturn(false);
int exceededTimeMs = 5;
time.sleep(DEFAULT_MAX_POLL_INTERVAL_MS + exceededTimeMs);
+ when(membershipManager.isLeavingGroup()).thenReturn(false);
NetworkClientDelegate.PollResult pollResult =
heartbeatRequestManager.poll(time.milliseconds());
assertEquals(1, pollResult.unsentRequests.size());
verify(membershipManager).transitionToSendingLeaveGroup(true);
verify(pollTimer, never()).isExpiredBy();
- assertEquals(exceededTimeMs, pollTimer.isExpiredBy());
clearInvocations(pollTimer);
heartbeatRequestManager.resetPollTimer(time.milliseconds());
verify(pollTimer).isExpiredBy();
}
- @Test
- public void testHeartbeatMetrics() {
- // setup
- coordinatorRequestManager = mock(CoordinatorRequestManager.class);
- membershipManager = mock(MembershipManager.class);
- heartbeatState = mock(HeartbeatRequestManager.HeartbeatState.class);
- time = new MockTime();
- metrics = new Metrics(time);
- heartbeatRequestState = new
HeartbeatRequestManager.HeartbeatRequestState(
- new LogContext(),
- time,
- 0, // This initial interval should be 0 to ensure heartbeat on the
clock
- DEFAULT_RETRY_BACKOFF_MS,
- DEFAULT_RETRY_BACKOFF_MAX_MS,
- 0);
- backgroundEventHandler = mock(BackgroundEventHandler.class);
- heartbeatRequestManager = createHeartbeatRequestManager(
- coordinatorRequestManager,
- membershipManager,
- heartbeatState,
- heartbeatRequestState,
- backgroundEventHandler);
-
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(new
Node(1, "localhost", 9999)));
- when(membershipManager.state()).thenReturn(MemberState.STABLE);
-
- assertNotNull(getMetric("heartbeat-response-time-max"));
- assertNotNull(getMetric("heartbeat-rate"));
- assertNotNull(getMetric("heartbeat-total"));
- assertNotNull(getMetric("last-heartbeat-seconds-ago"));
-
- // test poll
- assertHeartbeat(heartbeatRequestManager, 0);
- time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
- assertEquals(1.0, getMetric("heartbeat-total").metricValue());
- assertEquals((double)
TimeUnit.MILLISECONDS.toSeconds(DEFAULT_HEARTBEAT_INTERVAL_MS),
getMetric("last-heartbeat-seconds-ago").metricValue());
-
- assertHeartbeat(heartbeatRequestManager,
DEFAULT_HEARTBEAT_INTERVAL_MS);
- assertEquals(0.06d, (double)
getMetric("heartbeat-rate").metricValue(), 0.005d);
- assertEquals(2.0, getMetric("heartbeat-total").metricValue());
-
- // Randomly sleep for some time
- Random rand = new Random();
- int randomSleepS = rand.nextInt(11);
- time.sleep(randomSleepS * 1000);
- assertEquals((double) randomSleepS,
getMetric("last-heartbeat-seconds-ago").metricValue());
- }
-
@Test
public void
testFencedMemberStopHeartbeatUntilItReleasesAssignmentToRejoin() {
- mockStableMember();
+ heartbeatRequestManager = createHeartbeatRequestManager(
+ coordinatorRequestManager,
+ membershipManager,
+ heartbeatState,
+ heartbeatRequestState,
+ backgroundEventHandler);
time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
NetworkClientDelegate.PollResult result =
heartbeatRequestManager.poll(time.milliseconds());
assertEquals(1, result.unsentRequests.size());
// Receive HB response fencing member
when(subscriptions.hasAutoAssignedPartitions()).thenReturn(true);
- doNothing().when(membershipManager).transitionToFenced();
ClientResponse response =
createHeartbeatResponse(result.unsentRequests.get(0),
Errors.FENCED_MEMBER_EPOCH);
result.unsentRequests.get(0).handler().onComplete(response);
verify(membershipManager).transitionToFenced();
verify(heartbeatRequestState).onFailedAttempt(anyLong());
verify(heartbeatRequestState).reset();
+ when(membershipManager.shouldSkipHeartbeat()).thenReturn(true);
Review Comment:
nice, but this probably means that the following line expecting FENCED can
be removed.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
