mjsax commented on code in PR #23314:
URL: https://github.com/apache/kafka/pull/23314#discussion_r3994865074
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -163,14 +163,28 @@ private static void collectStandbyCandidates(final
Map<TaskId, ArrayList<Standby
}
}
+ // prev warm-up tasks: a warm-up task is a member restoring the state
of a task whose active task is being
+ // migrated to it, so for stickiness it counts as a prev standby --
the member that has already restored the
+ // state gets the task, instead of the restore work being thrown away.
Its reported offset sum ranks it among
+ // the standbys, most caught-up first.
+ for (final Map.Entry<String, Set<Integer>> entry :
memberAssignmentState.warmupTasks().entrySet()) {
+ final String subtopologyId = entry.getKey();
+ final Set<Integer> partitionNoSet = entry.getValue();
+ for (final int partitionNo : partitionNoSet) {
+ standbyCandidates
+ .computeIfAbsent(new TaskId(subtopologyId, partitionNo),
task -> new ArrayList<>())
+ .add(new StandbyCandidate(member, true,
reportedOffsetSum(memberAssignmentState, subtopologyId, partitionNo)));
Review Comment:
That's incorrect. The point of this PR is exactly to cover the case then
when old owner of the active task left the group (or was kicked out) and thus a
new owner must be assigned, and we want to consider warmup tasks for promotion
in addition to standby tasks.
--
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]
