mjsax commented on code in PR #22730:
URL: https://github.com/apache/kafka/pull/22730#discussion_r3584864076
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java:
##########
@@ -2286,6 +2543,58 @@ public void
testStreamsRebalanceDataHeartbeatIntervalMsUpdatedOnSuccess() {
}
}
+ @Test
+ public void testStreamsRebalanceDataTaskOffsetIntervalMsUpdatedOnSuccess()
{
+ try (final MockedConstruction<HeartbeatRequestState> ignored =
mockConstruction(
+ HeartbeatRequestState.class,
+ (mock, context) ->
when(mock.canSendRequest(time.milliseconds())).thenReturn(true)
+ )) {
+ final StreamsGroupHeartbeatRequestManager heartbeatRequestManager
= createStreamsGroupHeartbeatRequestManager();
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+ when(membershipManager.groupId()).thenReturn(GROUP_ID);
+ when(membershipManager.memberId()).thenReturn(MEMBER_ID);
+ when(membershipManager.memberEpoch()).thenReturn(MEMBER_EPOCH);
+
when(membershipManager.groupInstanceId()).thenReturn(Optional.of(INSTANCE_ID));
+
+ // The broker returns task.offset.interval.ms (KAFKA-18652) so the
client knows how often to report
+ // task changelog offsets; the client must store it in
StreamsRebalanceData.
+ assertEquals(-1, streamsRebalanceData.taskOffsetIntervalMs());
+
+ final NetworkClientDelegate.PollResult result =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, result.unsentRequests.size());
+
+
result.unsentRequests.get(0).handler().onComplete(buildClientResponse());
Review Comment:
Not sure what you mean, but I will remove this test anyway and fold into the
existing one.
--
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]
