gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4316872860
##########
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:
Step 2 still checks the previous members first, because a pick can have
looser conditions in step 2 than it had in step 1.
For example, with rack.aware.assignment.tags = `zone,cluster`: the task's
active is on M0 (zone1, cluster2) and its previous standby was on M1 (zone1,
cluster1). No process is in both a new zone and a new cluster, so in step 1 the
pick keeps only [zone], and its only candidate is M3 (zone2, cluster2). M1 is
not a candidate, so step 1 leaves this standby to step 2.
By the time step 2 picks for the task, standbys of other tasks have filled
M3. No process with room is in a new zone anymore, so the pick falls back to
[cluster], and M1 is a candidate again, together with M4 (zone1, cluster3). The
standby should go back to M1, which already has the state, rather than to M4,
even if M4 is less loaded.
The same can happen within one task: once step 2 places a standby, the next
pick for that task may drop a key whose values are now all used, and a previous
member can become a candidate.
--
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]
