lucasbru commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4025091550
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -288,6 +355,259 @@ 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
standing cap on how many replicas beyond
+ * the target assignment may exist at once.
+ *
+ * <p>Wherever a warm-up ends up it sits on the target owner
<em>itself</em>, so that it is promoted in place once
+ * it has caught up rather than closed on one member and reopened on
another. What varies is the price of getting
+ * it there:
+ * <ul>
+ * <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 slot.</li>
+ * <li>A <b>borrow</b> spends nothing, and applies when the target
owner <em>itself</em> already holds a
+ * standby of the task: that copy warms the migration as a side effect
of being a standby, and is the very
+ * replica the promotion then takes over in place. The target
assignment must be relocating it elsewhere --
+ * an assignor does not leave a standby on the process it hands the
active to -- so withholding that
+ * relocation keeps the replica count exactly where the target
assignment wants it. That withholding is the
+ * standby filter's job, not this one's.</li>
+ * <li>A standby on a <b>sibling</b> member of the target owner's
process competes for a slot like a fresh
+ * plant, because free warming is not on offer there. The sibling
cannot promote in place, so the copy has to
+ * move onto the target owner as a warm-up, and the redundancy it was
providing while it sat on the sibling
+ * then has to be backfilled by the relocation the target assignment
already wants -- three copies against
+ * the target assignment's two, which is one slot. Unfunded, it falls
back to being borrowed where it sits.</li>
+ * </ul>
+ *
+ * <p>Paying a slot for that move is worth it because a process holds a
task at most once, so the sibling has to
+ * release the task before the target owner can hold anything at all, and
only a store that persists to disk
+ * survives the release: the sibling's clean close leaves a checkpoint
behind for the incoming member to reopen
+ * from, whereas an in-memory store lives on the sibling's heap and is
dropped. Moving the copy onto the target
+ * owner pays that cost <em>during</em> the warming phase, where it merely
delays convergence. Leaving it on the
+ * sibling pays it at the hand-over instead, where it stalls processing --
which is the one thing staging exists
+ * to prevent.
+ *
+ * <p>A migration already being warmed keeps its slot ahead of any fresh
plant: dropping a restore part-way
+ * through to start another one elsewhere would throw away the very work
the budget exists to buy. What counts
+ * as already being warmed is a warm-up that a <em>still-staged</em>
migration justifies, which is why this
+ * reads the case analysis and not the current assignment. A task being
granted this step is no longer staged,
+ * so its warm-up is not kept and its slot is free again within this same
pass. The budget is recounted from
+ * zero on every call for the same reason: a warm-up whose task the
assignor has since re-targeted elsewhere
+ * must not go on holding a slot it no longer earns.
+ *
+ * <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 slot frees. Parking is never
destructive: no state is discarded
+ * because the budget ran out.
+ *
+ * @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 maxWarmupReplicas
+ * How many replicas 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 maxWarmupReplicas
+ ) {
+ // A budget of zero means the group does not stage migrations at all,
so there is nothing to fund. The
+ // caller short-circuits to the target assignment long before this,
which is where that contract lives --
+ // including that it disables the budget-free borrows too. This is
only the guard for a direct call.
+ if (maxWarmupReplicas == 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> keptWarmers = new ArrayList<>();
+ final List<FundingCandidate> plantCandidates = new ArrayList<>();
+
+ for (final StagedMigration migration : decisions.stagedMigrations()) {
+ switch (warmingOf(migration)) {
+ case PARK -> parkedMigrations.add(migration.task());
+ case BORROW -> borrowedMigrations.add(migration.task());
+ case KEEP -> keptWarmers.add(fundingCandidate(migration,
members, processLoad));
+ case PLANT -> plantCandidates.add(fundingCandidate(migration,
members, processLoad));
+ }
+ }
+
+ // Warm-ups in flight are funded first, but a budget that has shrunk
below their number -- a config change,
+ // since nothing else can lower it -- has to give some up. Evicting in
reverse funding order keeps which
+ // ones deterministic rather than dependent on iteration order.
+ keptWarmers.sort((left, right) -> comparePriority(left, right,
processLoad, Map.of()));
+ for (int i = 0; i < keptWarmers.size(); i++) {
+ final FundingCandidate keptWarmer = keptWarmers.get(i);
+ if (i < maxWarmupReplicas) {
+ warmupTasks.put(keptWarmer.task(), keptWarmer.targetOwner());
+ } else {
+ parkedMigrations.add(keptWarmer.task());
+ }
+ }
+
+ // Fresh plants take whatever the kept warmers left. Each one funded
raises its destination's load before
+ // the next pick, which spreads concurrent restores across processes
instead of stacking them all on
+ // whichever process happened to start out lightest -- so this picks
repeatedly rather than sorting once.
+ final Map<String, Integer> plantsByProcess = new HashMap<>();
+ int used = Math.min(keptWarmers.size(), maxWarmupReplicas);
+
+ while (used < maxWarmupReplicas && !plantCandidates.isEmpty()) {
+ int best = 0;
+ for (int candidate = 1; candidate < plantCandidates.size();
candidate++) {
+ final int comparison = comparePriority(
+ plantCandidates.get(candidate),
+ plantCandidates.get(best),
+ processLoad,
+ plantsByProcess
+ );
+ if (comparison < 0) {
+ best = candidate;
+ }
+ }
+
+ final FundingCandidate funded = plantCandidates.remove(best);
+ warmupTasks.put(funded.task(), funded.targetOwner());
+ plantsByProcess.merge(funded.targetProcessId(), 1, Integer::sum);
+ used++;
+ }
+
+ // A candidate that missed out is borrowed where it sits when the
target owner's process holds a standby on a
+ // sibling: warming through the sibling is worth more than not warming
at all, and is what the migration would
+ // have done anyway had the slot never been on offer. Everything else
has nothing to fall back on and parks.
Review Comment:
Sounds good. I think once we get to the optimization stage, we should make a
decision whether we need these extra collections or they are expendible,
because one obvious optimization would be to remove them.
--
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]
