bbejeck commented on code in PR #19691:
URL: https://github.com/apache/kafka/pull/19691#discussion_r2085303970
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1983,8 +1984,16 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
StreamsGroupHeartbeatResponseData response = new
StreamsGroupHeartbeatResponseData()
.setMemberId(updatedMember.memberId())
.setMemberEpoch(updatedMember.memberEpoch())
- .setHeartbeatIntervalMs(streamsGroupHeartbeatIntervalMs(groupId))
-
.setPartitionsByUserEndpoint(maybeBuildEndpointToPartitions(group));
+ .setHeartbeatIntervalMs(streamsGroupHeartbeatIntervalMs(groupId));
+
+ if (group.endpointInformationEpoch() > memberEndpointEpochInRequest) {
+
response.setPartitionsByUserEndpoint(maybeBuildEndpointToPartitions(group));
+
response.setEndpointInformationEpoch(group.endpointInformationEpoch());
+ } else {
+ int responseEndpoint = Math.min(group.endpointInformationEpoch(),
memberEndpointEpochInRequest);
Review Comment:
We need to account for the `GroupMetadataManager` losing the state of the
endpoint epoch, in that case we'll reset the clients to use the GMM endpoint
epoch.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1993,6 +2002,7 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
response.setActiveTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().activeTasks()));
response.setStandbyTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().standbyTasks()));
response.setWarmupTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().warmupTasks()));
+ group.setEndpointInformationEpoch(group.endpointInformationEpoch()
+ 1);
Review Comment:
Bump the endpoint information epoch here as there's been assignment changes.
The endpoint information will be sent in the next heartbeat response. We'll
do it this way, as the next successful heartbeat request/response cycle the
group member will have accepted the new assignment.
--
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
