lucasbru commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4153545382
##########
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;
Review Comment:
leastLoadedMemberWithRoom scans every member of the process, and it's called
here for every candidate on every pickCandidates() filter pass (could be more
than once per standby, if a key gets given up), then again at placement time
for the winner. With membersPerProcess around 50 like the benchmark uses,
that's a lot of repeat scanning. Could we cache the
least-loaded-member-with-room per process and only invalidate it when that
process actually gets a placement?
##########
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:
This prev-active -> prev-standby-least-loaded -> least-loaded tie-break is
the same one already written for the tag-blind path a bit above. Worth pulling
into a shared helper so the two passes can't drift apart if the tie-break rule
ever changes.
--
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]
