brenden20 commented on code in PR #16200:
URL: https://github.com/apache/kafka/pull/16200#discussion_r1649440775
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -106,72 +102,99 @@ public class HeartbeatRequestManagerTest {
private MembershipManager membershipManager;
private HeartbeatRequestManager.HeartbeatRequestState
heartbeatRequestState;
private HeartbeatRequestManager.HeartbeatState heartbeatState;
- private final String memberId = "member-id";
- private final int memberEpoch = 1;
private BackgroundEventHandler backgroundEventHandler;
- private Metrics metrics;
+ private LogContext logContext;
@BeforeEach
public void setUp() {
- setUp(ConsumerTestBuilder.createDefaultGroupInformation());
- }
-
- private void setUp(Optional<ConsumerTestBuilder.GroupInformation>
groupInfo) {
- testBuilder = new ConsumerTestBuilder(groupInfo, true, false);
- time = testBuilder.time;
- coordinatorRequestManager =
testBuilder.coordinatorRequestManager.orElseThrow(IllegalStateException::new);
- heartbeatRequestManager =
testBuilder.heartbeatRequestManager.orElseThrow(IllegalStateException::new);
- heartbeatRequestState =
testBuilder.heartbeatRequestState.orElseThrow(IllegalStateException::new);
- heartbeatState =
testBuilder.heartbeatState.orElseThrow(IllegalStateException::new);
- backgroundEventHandler = testBuilder.backgroundEventHandler;
- subscriptions = testBuilder.subscriptions;
- membershipManager =
testBuilder.membershipManager.orElseThrow(IllegalStateException::new);
- metadata = testBuilder.metadata;
- metrics = new Metrics(time);
+ this.time = new MockTime();
+ Metrics metrics = new Metrics(time);
+ this.logContext = new LogContext();
+ this.pollTimer = mock(Timer.class);
+ this.coordinatorRequestManager = mock(CoordinatorRequestManager.class);
+ this.heartbeatState = mock(HeartbeatState.class);
+ this.backgroundEventHandler = mock(BackgroundEventHandler.class);
+ this.subscriptions = mock(SubscriptionState.class);
+ this.membershipManager = mock(MembershipManagerImpl.class);
+ this.metadata = mock(ConsumerMetadata.class);
+ ConsumerConfig config = mock(ConsumerConfig.class);
+
+ this.heartbeatRequestState = spy(new HeartbeatRequestState(
+ logContext,
+ time,
+ DEFAULT_HEARTBEAT_INTERVAL_MS,
+ DEFAULT_RETRY_BACKOFF_MS,
+ DEFAULT_RETRY_BACKOFF_MAX_MS,
+ DEFAULT_HEARTBEAT_JITTER_MS));
+
+ this.heartbeatRequestManager = new HeartbeatRequestManager(
+ logContext,
+ pollTimer,
+ config,
+ coordinatorRequestManager,
+ membershipManager,
+ heartbeatState,
+ heartbeatRequestState,
+ backgroundEventHandler,
+ metrics);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(new
Node(1, "localhost", 9999)));
+ Map<Uuid, SortedSet<Integer>> map = new HashMap<>();
+ LocalAssignment local = new LocalAssignment(0, map);
+ when(membershipManager.currentAssignment()).thenReturn(local);
}
- private void resetWithZeroHeartbeatInterval(Optional<String>
groupInstanceId) {
- cleanup();
-
- ConsumerTestBuilder.GroupInformation gi = new
ConsumerTestBuilder.GroupInformation(
- DEFAULT_GROUP_ID,
- groupInstanceId,
+ private void createHeartbeatStateWith0HeartbeatInterval() {
+ this.heartbeatRequestState = spy(new HeartbeatRequestState(
+ logContext,
+ time,
0,
- 0.0,
- Optional.of(DEFAULT_REMOTE_ASSIGNOR)
- );
+ DEFAULT_RETRY_BACKOFF_MS,
+ DEFAULT_RETRY_BACKOFF_MAX_MS,
+ DEFAULT_HEARTBEAT_JITTER_MS));
- setUp(Optional.of(gi));
+ heartbeatRequestManager = createHeartbeatRequestManager(
+ coordinatorRequestManager,
+ membershipManager,
+ heartbeatState,
+ heartbeatRequestState,
+ backgroundEventHandler);
}
- @AfterEach
- public void cleanup() {
- if (testBuilder != null) {
- testBuilder.close();
- }
+ private void resetWithZeroHeartbeatInterval() {
Review Comment:
It works for all but one test, for now I am keeping
```createHeartbeatStateWith0HeartbeatInterval()``` for
```testSkippingHeartbeat(final boolean shouldSkipHeartbeat)``` since this test
requires the initial heartbeatInterval be 0
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -275,29 +316,34 @@ public void testHeartbeatNotSentIfAnotherOneInFlight() {
assertEquals(1, result.unsentRequests.size());
NetworkClientDelegate.UnsentRequest inflightReq =
result.unsentRequests.get(0);
+ // Receive response for the inflight after the interval expired. The
next HB should be sent
+ // on the next poll waiting only for the minimal backoff.
+ inflightReq.handler().onComplete(createHeartbeatResponse(inflightReq,
Errors.NONE));
+ time.sleep(DEFAULT_RETRY_BACKOFF_MS);
+ when(heartbeatRequestState.canSendRequest(anyLong())).thenReturn(true);
Review Comment:
Looks like at some point I changed the order, switched it back and it is
working good now
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -744,34 +798,17 @@ public void
testFencedMemberStopHeartbeatUntilItReleasesAssignmentToRejoin() {
verify(heartbeatRequestState).onFailedAttempt(anyLong());
verify(heartbeatRequestState).reset();
+
when(heartbeatRequestState.canSendRequest(anyLong())).thenReturn(false);
when(membershipManager.state()).thenReturn(MemberState.FENCED);
result = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, result.unsentRequests.size(), "Member should not send
heartbeats while FENCED");
+ when(heartbeatRequestState.canSendRequest(anyLong())).thenReturn(true);
when(membershipManager.state()).thenReturn(MemberState.JOINING);
Review Comment:
Good catch there, thank you!
--
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
