mjsax commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4018729548
##########
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.
+ plantCandidates.forEach(candidate -> {
+ if (candidate.borrowable()) {
+ borrowedMigrations.add(candidate.task());
+ } else {
+ parkedMigrations.add(candidate.task());
+ }
+ });
+
+ return new WarmupPlan(
+ Collections.unmodifiableSortedMap(warmupTasks),
+ Collections.unmodifiableSortedSet(borrowedMigrations),
+ Collections.unmodifiableSortedSet(parkedMigrations)
+ );
+ }
+
+ /**
+ * Resolves the parts of a staged migration the funding order needs, once,
so that the repeated comparisons do
+ * not each redo the lookups.
+ *
+ * <p>The source load is resolved here rather than compared lazily because
it cannot change during the pass:
+ * funding a warm-up adds a task to its <em>destination</em> process,
while the source keeps running the active
+ * task either way.
+ */
+ private static FundingCandidate fundingCandidate(
+ final StagedMigration migration,
+ final Map<String, StreamsGroupMember> members,
+ final Map<String, ProcessLoad> processLoad
+ ) {
+ final String sourceProcessId =
members.get(migration.currentOwner()).processId();
+ return new FundingCandidate(
+ migration.task(),
+ migration.targetOwner(),
+ migration.targetProcessId().orElseThrow(),
+ processLoad.get(sourceProcessId).load(),
+ isBorrowableFromSibling(migration)
+ );
+ }
+
+ /**
+ * How a staged migration is to be warmed, which is decided entirely by
what the target owner's process already
+ * holds for the task -- and, when it holds a standby, by whether that
standby sits on the target owner itself.
+ */
+ private static Warming warmingOf(final StagedMigration migration) {
+ if (migration.targetProcessId().isEmpty()) {
+ // The target assignment names a member the group no longer has,
so there is nowhere to warm up and no
+ // slot may be spent. The task simply stays with its current owner.
+ return Warming.PARK;
+ }
+
+ final Optional<TaskCopy> copyOnTargetProcess =
migration.copyOnTargetProcess();
+ if (copyOnTargetProcess.isEmpty()) {
+ return Warming.PLANT;
+ }
+ if (copyOnTargetProcess.get().role() == TaskRole.WARMUP) {
+ return Warming.KEEP;
+ }
+ // A standby on the target owner itself is borrowed outright, since
the promotion takes it over in place. One
+ // on a sibling warms nothing the promotion can take over, so it has
to move onto the target owner, and that
+ // competes for a slot to pay for the redundancy backfill which
follows it across.
+ return
copyOnTargetProcess.get().memberId().equals(migration.targetOwner())
+ ? Warming.BORROW
+ : Warming.PLANT;
Review Comment:
> Why do you use ? : here instead of if like everywhere above.
That was Claude -- I can change it -- personally, I don't care either way.
> Why is the "else" branch PLANT here. I cannot put a warm-up task on that
process becuase it would conflict with the standby task. Would it make sense to
add a separate warming classification for this case? What do we do, remove the
standby?
Yes, but this all happens automatically. We know, that that target
assignment puts the active task on a member of this process. This implies, the
target assignment also removes the existing standby from this process
(otherwise the assignor would be broken, and we don't want to make the refiner
responsible to verify if the assignor computed something incorrect [I am open
to add a check for the assignor, given that we support custom assignors, but
such a check should be done right after we get the assignment back, and before
we call the refiner]). Thus, we can just assign warmup directly. The reconciler
will take care of the rest for us (ie, revoke the standby on the sibling first,
before it can assign the warmup on the target owner).
The key is, that we are patching the target assignment, so for the patch-set
we need to keep the active on keep it on the old owner, and add a warmup to the
target owner. The standby is already correct in the target assignment and
doesn't need patching.
--
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]
