gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4174593162
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -524,6 +605,123 @@ 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<ProcessGroup> picker,
+ final TaskId task,
+ final List<ProcessState> alreadyPlaced,
+ final boolean stickyOnly,
+ final List<TaskId> rackNonSticky
+ ) {
+ picker.startTask();
+
picker.markUsed(localState.processIdToGroup.get(localState.statefulActiveTaskToProcess.get(task).processId()));
+ for (final ProcessState process : alreadyPlaced) {
+
picker.markUsed(localState.processIdToGroup.get(process.processId()));
+ }
+
+ // The picker only tests groups whose value for the priority key no
holder carries, so no process of such a group
+ // holds the task and only room is checked.
+ final Predicate<ProcessGroup> eligible = group ->
group.hasRoom(localState);
+
+ final List<ProcessState> placed = new ArrayList<>();
+ while (alreadyPlaced.size() + placed.size() <
localState.numStandbyReplicas) {
+ final Set<ProcessGroup> candidates =
picker.pickCandidates(eligible);
+ if (candidates.isEmpty()) {
+ break;
+ }
+
+ // prev active or standby member on a process with room in a
candidate group
+ final Member prevMember = findPrevMemberForStandby(
+ localState,
+ task,
+ process ->
candidates.contains(localState.processIdToGroup.get(process.processId())) &&
leastLoadedMemberWithRoom(localState, process) != null
+ );
+ if (prevMember != null) {
+ placeRackAwareStandby(localState, picker,
localState.processIdToState.get(prevMember.processId), prevMember.memberId,
task, placed);
+ continue;
+ }
+
+ if (stickyOnly) {
+ rackNonSticky.add(task);
+ break;
+ }
+
+ // least loaded process of the candidate groups. It has a member
with room, so the lookup cannot return null.
+ // Adding to that member directly keeps the heap of members by
load out of this pass, so that it is built
+ // once, in the least-loaded pass, as without rack awareness.
+ final ProcessState processWithLeastLoad = leastLoaded(localState,
candidates);
+ placeRackAwareStandby(localState, picker, processWithLeastLoad,
leastLoadedMemberWithRoom(localState, processWithLeastLoad), task, placed);
+ }
+ return placed;
+ }
+
+ private static void placeRackAwareStandby(
+ final LocalState localState,
+ final RackAwareStandbyPicker<ProcessGroup> picker,
+ final ProcessState process,
+ final String memberId,
+ final TaskId task,
+ final List<ProcessState> placed
+ ) {
+ // The queue of the group catches up with the new load once the
process reaches its head.
+ maybeUpdateTotalTasksPerMember(localState, process.addTask(memberId,
task, false, true));
+ picker.markUsed(localState.processIdToGroup.get(process.processId()));
+ placed.add(process);
+ }
+
+ /** The least-loaded process with room of the candidate groups, which all
have one. */
+ private static ProcessState leastLoaded(final LocalState localState, final
Collection<ProcessGroup> candidates) {
Review Comment:
cause for rack-aware pick, the processes for leastLoaded() are those process
also satisfy rack-diverse condition. So it won't save that much of time since
each time the input are different. Also, since we are using group, the
O(#surviver_groups) should be small enough
--
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]
