mjsax commented on code in PR #22748:
URL: https://github.com/apache/kafka/pull/22748#discussion_r3555716562
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2196,21 +2201,29 @@ 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;
+ }
+
+ // Maybe check if warmup task are hot, and the assignment can be
refined furtther
+ if (assignmentUpdate == AssignmentUpdate.NONE
+ && group.state() == StreamsGroup.StreamsGroupState.STABLE
+ && hasHotWarmupTask(group,
streamsGroupAcceptableRecoveryLag(groupId))
+ ) {
+ assignmentUpdate = AssignmentUpdate.REFINE;
}
// Actually bump the group epoch
int groupEpoch = group.groupEpoch();
- if (bumpGroupEpoch) {
+ if (assignmentUpdate != AssignmentUpdate.NONE) {
Review Comment:
Fair point. Not sure right now how we could improve the control flow...
If we have hot warmup tasks, we _expect_ the refiner to revoke the
corresponding active tasks, and thus be bump the group-epoch pro-actively. But
I see your point that it scattered logic, and could become a "split brain"
problem. Let me think about it...
--
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]
