mjsax commented on code in PR #22748:
URL: https://github.com/apache/kafka/pull/22748#discussion_r3640709899
##########
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;
+ }
+
+ TasksTuple refinedAssignment = null;
+ if (assignmentUpdate == AssignmentUpdate.NONE && group.state() ==
StreamsGroup.StreamsGroupState.STABLE) {
+ // We are not computing a new target assignment later, thus we try
to refine the current target assignment
+ // into an intermediate assignment (with warm-up tasks) the member
should be reconciled towards.
+ refinedAssignment = maybeRefineAssignment( // no-op for now
+ updatedMember,
+ group.targetAssignment(),
+ group.taskOffsets(),
+ streamsGroupNumWarmupReplicas(groupId),
+ streamsGroupAcceptableRecoveryLag(groupId)
+ );
Review Comment:
My understanding it, we would be in state RECONCILING, and never this the
`refine(...)` step here, as we gate on `STABLE` above.
--
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]
