squah-confluent commented on code in PR #22748:
URL: https://github.com/apache/kafka/pull/22748#discussion_r3637775335
##########
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:
What happens here when the heartbeating member is a newly rejoined static
member and `group.targetAssignment()` is the un-updated map?
##########
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:
Unrelated to this PR - do we have a bug here when there is a deferred
assignment? It looks like we might bump the epoch on every heartbeat when there
is a pending assignment config change.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4358,6 +4421,14 @@ private UpdateTargetAssignmentResult<Map<String,
TasksTuple>> maybeUpdateStreams
return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(),
updatedMembersAndTargetAssignment.targetAssignment());
}
+ if (refineOnly) {
Review Comment:
I'm also confused by @dajac's comment.
--
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]
