ChoMinGi commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4033920273
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -288,6 +286,240 @@ private static boolean isReady(
.anyMatch(holder -> holder.processId().equals(targetProcessId) &&
holder.caughtUp());
}
+ /**
+ * Decides which of the staged migrations get a warm-up task, under the
warmup budget.
+ *
+ * <p>There is different scenarios:
+ * <ul>
+ * <li>A warm-up task already restoring keeps its warm-up slot if the
target assignment didn't change, and the
+ * warmu-up task is not caught up yet. It could also get revoked if
the warmup budget was reduced and keeping
+ * the warmup would now exceed the budget.
+ * <li>A <b>fresh plant</b> puts a warm-up task on a target owner
whose process holds nothing for the task,
+ * and spends a warm-up slot.</li>
+ * <li>When the target owner <em>itself</em> already holds a standby
of the task we can <b>borrow</b> it,
+ * and no warmup budget is used: that standby warms-up the task
anyway.</li>
+ * <li>If a target member's <em>sibling</em> hold a standby, we cannot
borrow but, but need to move the
+ * standby to its new owner, and putting a warmup on the target
member, spending a warm-up slot.
+ * (Cf case (2) of {@link #isReady(CurrentAssignmentIndex, TaskId,
String, String)} </li>
+ * </ul>
+ *
+ * <p>Everything else <b>parks</b> -- the task keeps running on its
current owner with nothing warming up, and a
+ * later refinement step picks it up once a warm-up slot frees.
+ *
+ * @param decisions
+ * What the case analysis decided, from {@link #analyzeTasks}.
+ * @param members
+ * All members of the group, used to resolve which process a task's
current owner runs in.
+ * @param processLoad
+ * The load of each process, from {@link #indexProcessLoad}.
+ * @param numWarmupReplicas
+ * How many copies beyond the target assignment may exist at once,
group-wide.
+ *
+ * @return Which warm-up tasks the intermediate assignment places, and how
each staged migration is warmed.
+ */
+ static WarmupPlan planWarmups(
+ final TaskDecisions decisions,
+ final Map<String, StreamsGroupMember> members,
+ final Map<String, ProcessLoad> processLoad,
+ final int numWarmupReplicas
+ ) {
+ if (numWarmupReplicas == 0) {
+ return WarmupPlan.EMPTY;
+ }
+
+ final SortedMap<TaskId, String> warmupTasks = new TreeMap<>();
+ final SortedSet<TaskId> borrowedMigrations = new TreeSet<>();
+ final SortedSet<TaskId> parkedMigrations = new TreeSet<>();
+
+ final List<FundingCandidate> keptWarmups = new ArrayList<>();
+ final List<FundingCandidate> newWarmupCandidates = new ArrayList<>();
+
+ for (final StagedMigration migration : decisions.stagedMigrations()) {
+ final Warming warming = warmingOf(migration);
+ switch (warming) {
+ case PARK -> parkedMigrations.add(migration.task());
+ case BORROW -> borrowedMigrations.add(migration.task());
+ case KEEP -> keptWarmups.add(fundingCandidate(migration,
members, warming));
+ // Both put a warm-up task on the target owner and both cost a
warm-up slot, so they share one
+ // candidate list -- but a plant is funded ahead of a sibling
move (see comparePriority).
+ case PLANT, SIBLING_MOVE ->
+ newWarmupCandidates.add(fundingCandidate(migration,
members, warming));
+ }
+ }
+
+ // Warm-up tasks already restoring are funded first. If {@code
num.warmup.replicas} config was reduced, we might
+ // be over warmup budget and have to give up some warmup tasks.
Evicting in reverse funding order
+ // keeps which ones deterministic rather than dependent on iteration
order.
+ // Note: revocation of warmup task happens implicitly by not adding
them to the assignment patch again
+ keptWarmups.sort((left, right) -> comparePriority(left, right,
processLoad, Map.of()));
+ for (int i = 0; i < keptWarmups.size(); i++) {
+ final FundingCandidate keptWarmup = keptWarmups.get(i);
+ if (i < numWarmupReplicas) {
+ warmupTasks.put(keptWarmup.task(), keptWarmup.targetOwner());
+ } else {
+ parkedMigrations.add(keptWarmup.task());
+ }
+ }
+
+ // New warm-up tasks raises its target process's load, so we need to
update it while we go, and find a new
+ // `best` from scratch each time
+ // note: this nested-loop is bounded by the number of unused warm-up
slots; so while it's O(unused * candidate)
+ // it's effectively not quadratic (we can consider `unused` a constant)
+ final Map<String, Integer> newWarmupsByProcess = new HashMap<>();
+ int used = Math.min(keptWarmups.size(), numWarmupReplicas);
+
+ while (used < numWarmupReplicas && !newWarmupCandidates.isEmpty()) {
+ int best = 0;
+ for (int candidate = 1; candidate < newWarmupCandidates.size();
candidate++) {
+ final int comparison = comparePriority(
+ newWarmupCandidates.get(candidate),
+ newWarmupCandidates.get(best),
+ processLoad,
+ newWarmupsByProcess
+ );
+ if (comparison < 0) {
+ best = candidate;
+ }
+ }
+
+ final FundingCandidate funded = newWarmupCandidates.remove(best);
+ warmupTasks.put(funded.task(), funded.targetOwner());
+ newWarmupsByProcess.merge(funded.targetProcessId(), 1,
Integer::sum);
Review Comment:
For `SIBLING_MOVE`, indexProcessLoad() already counts the standby on the
target process. Since the warm-up replaces it on the same process, shouldn't
the process load stay unchanged?
Should this increment apply only to `PLANT`?
--
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]
