gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4157657522
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -524,6 +580,122 @@ private static void assignStandby(final LocalState
localState, final LinkedList<
}
}
+ /**
+ * Places the standbys of {@code task} that still make it more
rack-diverse on processes with room and returns
+ * the processes they went to. Among the equally most diverse processes
the choice is the tag-blind picks unchanged:
+ * the previous active member, else the least-loaded previous standby
member, each while below the quota, else
+ * the least-loaded process and its least-loaded member.
+ * <p>
+ * {@code alreadyPlaced} are the standbys of the task placed by an earlier
call. With {@code stickyOnly},
+ * the picks stop at the first standby that would not go back to a
previous member and add the task to
+ * {@code rackNonSticky}, leaving it and the rest to a later call.
+ */
+ private static List<ProcessState> pickRackAwareStandbys(
+ final LocalState localState,
+ final RackAwareStandbyPicker<ProcessState> picker,
+ final TaskId task,
+ final List<ProcessState> alreadyPlaced,
+ final boolean stickyOnly,
+ final List<TaskId> rackNonSticky
+ ) {
+ picker.startTask();
+ picker.markUsed(localState.statefulActiveTaskToProcess.get(task));
+ for (final ProcessState process : alreadyPlaced) {
+ picker.markUsed(process);
+ }
+
+ final Predicate<ProcessState> eligible =
+ process -> !process.hasTask(task) &&
leastLoadedMemberWithRoom(localState, process) != null;
+
+ final List<ProcessState> placed = new ArrayList<>();
+ while (alreadyPlaced.size() + placed.size() <
localState.numStandbyReplicas) {
+ final Set<ProcessState> candidates =
picker.pickCandidates(eligible);
+ if (candidates.isEmpty()) {
+ break;
+ }
+
+ // prev active task, if it is a candidate
Review Comment:
sounds good, I will turn it into a helper.
--
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]
