gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4316914615
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -525,6 +564,100 @@ private static void assignStandby(final LocalState
localState, final LinkedList<
}
}
+ /**
+ * Steps 1 and 2 of {@link #assignStandby}: places the standbys that make
their task more rack-diverse and returns
+ * the processes they went to per task.
+ */
+ private static Map<TaskId, List<ProcessState>>
assignRackAwareStandbys(final LocalState localState, final List<TaskId>
standbyTasks) {
+ // A process has room while one of its members is below the per-member
quota. Loads and room change only when a
+ // standby is added below, and each change is reported to the picker.
+ final RackAwareStandbyPicker<ProcessState> picker = new
RackAwareStandbyPicker<>(
+ localState.rackAwareAssignmentTags,
+ localState.processIdToState.values(),
+ process ->
localState.processIdToClientTags.get(process.processId()),
+ ProcessState::load,
+ process -> hasRoom(localState, process)
+ );
+ final Map<TaskId, List<ProcessState>> rackAwareStandbys = new
HashMap<>();
+
+ // 1. re-assigning standby tasks in a new rack to clients that
previously had the same task (as active or standby)
+ for (final TaskId task : standbyTasks) {
+ final List<ProcessState> standbys = new ArrayList<>();
+ rackAwareStandbys.put(task, standbys);
+ placeRackAwareStandbys(localState, picker, task, standbys, true);
+ }
+
+ // 2. assigning remaining standby tasks in a new rack to the least
loaded client
+ for (final TaskId task : standbyTasks) {
+ placeRackAwareStandbys(localState, picker, task,
rackAwareStandbys.get(task), false);
+ }
+ return rackAwareStandbys;
+ }
+
+ /**
+ * Places standbys of {@code task} while they make it more rack-diverse
and adds their processes to {@code standbys},
+ * which holds those placed by an earlier call. Among the candidates of a
pick, a standby goes to the previous active
+ * member, else the least-loaded previous standby member, each while below
the quota, else to the least-loaded
+ * candidate. With {@code stickyOnly}, stops at the first standby with no
previous member to go to.
+ */
+ private static void placeRackAwareStandbys(
+ final LocalState localState,
+ final RackAwareStandbyPicker<ProcessState> picker,
+ final TaskId task,
+ final List<ProcessState> standbys,
+ final boolean stickyOnly
+ ) {
+ final List<ProcessState> holders = new ArrayList<>(1 +
standbys.size());
+ holders.add(localState.statefulActiveTaskToProcess.get(task));
+ holders.addAll(standbys);
+ picker.startTask(holders);
+
+ while (standbys.size() < localState.numStandbyReplicas) {
+ final RackAwareStandbyPicker.Pick<ProcessState> pick =
picker.pick();
+ if (pick == null) {
+ return;
+ }
+
+ // prev active or standby member on a candidate process, else the
least-loaded candidate
+ final Member prevMember = findPrevMemberForStandby(localState,
task, candidate -> picker.isCandidate(candidate, pick.conditions()));
+ final ProcessState process;
+ final int newTaskCount;
+ if (prevMember != null) {
Review Comment:
while the metric in rack aware basically the same, the stickiness increased
--
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]
