lianetm commented on code in PR #22730:
URL: https://github.com/apache/kafka/pull/22730#discussion_r3579417132
##########
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:
isn't this the same that completeSuccessfulHeartbeart does?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -650,9 +650,24 @@ private void onSuccessResponse(final
StreamsGroupHeartbeatResponse response, fin
heartbeatRequestState.updateHeartbeatIntervalMs(data.heartbeatIntervalMs());
heartbeatRequestState.onSuccessfulAttempt(currentTimeMs);
heartbeatState.setEndpointInformationEpoch(data.endpointInformationEpoch());
-
streamsRebalanceData.setHeartbeatIntervalMs(data.heartbeatIntervalMs());
-
streamsRebalanceData.setTaskOffsetIntervalMs(data.taskOffsetIntervalMs());
-
streamsRebalanceData.setAcceptableRecoveryLag(data.acceptableRecoveryLag());
+ // A leaving member's response carries no group configuration (the
fields hold protocol defaults), so do not
+ // log or store it while shutting down. Normal responses have
memberEpoch >= 0; leave responses use the
+ // negative leave sentinels (fenced members take the error path
instead). Log only when a value changes, to
+ // avoid repeating it on every heartbeat: this fires on first receipt
(values start unset) and on any later change.
+ if (data.memberEpoch() >= 0) {
Review Comment:
interesting, this is a behaviour change really, more than logs. I agree with
it, just wondering if we should call it out in the PR title or body? With this
we're preserving streams configs received from the GC after the member leaves.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java:
##########
@@ -2286,6 +2543,58 @@ public void
testStreamsRebalanceDataHeartbeatIntervalMsUpdatedOnSuccess() {
}
}
+ @Test
+ public void testStreamsRebalanceDataTaskOffsetIntervalMsUpdatedOnSuccess()
{
Review Comment:
aren't these 2 + the hearbeatInterval one almost the same, just checking
different fields? If so we maybe have a single test that asserts the 3 fields
instead?
--
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]
