mjsax commented on code in PR #22748:
URL: https://github.com/apache/kafka/pull/22748#discussion_r3640760187
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2201,21 +2205,37 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
// We validated a topology that was not validated before, so bump the
group epoch as we may have to reassign tasks.
if (validatedTopologyEpoch != group.validatedTopologyEpoch()) {
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
}
// Check if assignment configurations have changed
Map<String, String> currentAssignmentConfigs =
streamsGroupAssignmentConfigs(groupId);
Map<String, String> storedAssignmentConfigs =
group.lastAssignmentConfigs();
- if (!bumpGroupEpoch &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
+ if (assignmentUpdate == AssignmentUpdate.NONE &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
log.info("[GroupId {}][MemberId {}] Assignment configurations
changed to {}. Triggering rebalance.",
groupId, memberId, currentAssignmentConfigs);
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
+ }
Review Comment:
I don't think it's a bug -- we do record the config change independent of
the assignment run, ie, it's recoded even if the assignment run is deferred --
thus, the next heartbeat would see matching configs.
We don't have a test for this though -- let me add on in a follow up PR
--
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]
