Copilot commented on code in PR #20854:
URL: https://github.com/apache/kafka/pull/20854#discussion_r2510897439
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2092,15 +2092,21 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
response.setActiveTasks(createStreamsGroupHeartbeatResponseTaskIdsFromEpochs(updatedMember.assignedTasks().activeTasksWithEpochs()));
response.setStandbyTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().standbyTasks()));
response.setWarmupTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().warmupTasks()));
- if (memberEpoch != 0 || !updatedMember.assignedTasks().isEmpty()) {
+ if (updatedMember.userEndpoint().isPresent()) {
+ // If no user endpoint is defined, there is no change in the
endpoint information.
+ // Otherwise, bump the endpoint information epoch
group.setEndpointInformationEpoch(group.endpointInformationEpoch() + 1);
}
}
if (group.endpointInformationEpoch() != memberEndpointEpoch) {
response.setPartitionsByUserEndpoint(maybeBuildEndpointToPartitions(group,
updatedMember));
}
- response.setEndpointInformationEpoch(group.endpointInformationEpoch());
+ if (groups.containsKey(group.groupId())) {
+ // If we just created the group and the endpoint information epoch
will not be persisted, so return epoch 0.
Review Comment:
The comment on line 2106 has grammatical issues. It should read: 'If we just
created the group, the endpoint information epoch will not be persisted, so
return epoch 0.' Alternatively, rephrase to: 'If the group was just created,
the endpoint information epoch will not yet be persisted, so we omit it
(implicitly returning 0).'
```suggestion
// If we just created the group, the endpoint information epoch
will not be persisted, so return epoch 0.
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2092,15 +2092,21 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
response.setActiveTasks(createStreamsGroupHeartbeatResponseTaskIdsFromEpochs(updatedMember.assignedTasks().activeTasksWithEpochs()));
response.setStandbyTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().standbyTasks()));
response.setWarmupTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().warmupTasks()));
- if (memberEpoch != 0 || !updatedMember.assignedTasks().isEmpty()) {
+ if (updatedMember.userEndpoint().isPresent()) {
+ // If no user endpoint is defined, there is no change in the
endpoint information.
+ // Otherwise, bump the endpoint information epoch
group.setEndpointInformationEpoch(group.endpointInformationEpoch() + 1);
}
}
if (group.endpointInformationEpoch() != memberEndpointEpoch) {
response.setPartitionsByUserEndpoint(maybeBuildEndpointToPartitions(group,
updatedMember));
}
- response.setEndpointInformationEpoch(group.endpointInformationEpoch());
+ if (groups.containsKey(group.groupId())) {
+ // If we just created the group and the endpoint information epoch
will not be persisted, so return epoch 0.
+ // Otherwise, return the bumped epoch.
Review Comment:
The logic appears inverted. The condition
`groups.containsKey(group.groupId())` checks if the group exists in the
persisted map, but the comment suggests setting the epoch when the group is NOT
yet persisted. If the group is newly created via `getOrCreateStreamsGroup`, it
won't be in the `groups` map until replay. Consider clarifying the logic or
inverting the condition to match the comment's intent: when the group is NOT in
the map (newly created), we should skip setting the epoch; when it IS in the
map (persisted), we should set it.
```suggestion
// If we just created the group and the endpoint information epoch
will not be persisted, so return epoch 0.
// Otherwise, return the bumped epoch.
if (!groups.containsKey(group.groupId())) {
response.setEndpointInformationEpoch(0);
} else {
```
--
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]
