mjsax commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4018509016
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -268,12 +323,24 @@ static TaskDecisions analyzeTasks(
* is dropped when the task closes, and no hand-over of a running task
between threads of one process exists to
* carry it across. Worse, the lag that made the task look ready was
measured on the member that is about to
* close, so for an in-memory store it says nothing about what the
incoming member then has to restore. This
- * predicate cannot fix that; it would take a client-side cross-thread
task hand-over. The broker cannot even see
- * the difference, because the topology metadata carries changelog topics
but not how a store is backed.
+ * predicate cannot fix that; it would take a client-side cross-thread
task hand-over
+ * (https://issues.apache.org/jira/browse/KAFKA-21090). The broker cannot
even see the difference, because the
+ * topology metadata carries changelog topics but not how a store is
backed.
+ *
+ * <p>The damage is bounded, though, because <b>the refiner never creates
one of those two paths -- it only ever
+ * inherits them.</b> Every warm-up it plants sits on the target owner
itself, so every migration it warms ends in
+ * the in-place promotion, which is warm for every store type. It even
pays to keep that true: where the
+ * destination process holds a copy of the task only on a <em>sibling</em>
of the target owner, the budget pass
+ * spends a slot to move that copy onto the target owner rather than
borrow it where it sits. So the only way to
+ * reach one of the two cold paths is through this predicate granting the
task outright -- nothing was warmed, and
+ * the layout was already there when the refiner looked.
*
- * <p>What bounds the damage is that a warm-up task the refiner plants
always targets the target owner itself, so
- * every migration the refiner stages resolves through the in-place
promotion. The other paths arise only out of a
- * layout the refiner inherited.
+ * <p>Note that includes a copy on a sibling member that is
<em>already</em> caught up: the task is granted here,
+ * in this step, before the budget pass ever sees the migration, so
nothing gets the chance to move that copy onto
+ * the target owner first. Doing so would spend a slot to buy an in-place
promotion -- which is worth it for an
+ * in-memory store and pure waste for a store that persists to disk, since
that one reopens warm from the state
+ * directory anyway. The broker cannot tell the two apart, so this grants
immediately and converges fast. It is a
+ * deliberate boundary rather than an oversight, and the design document
carries it as an open question.
Review Comment:
It's a synonym for placing a warmup task... if we have config
`num.warmup.replicas = 2` we have two warmup slots... I think it's a good term.
Let me know if I should change it (and if yes: any suggestions)? We pay for a
slot when we decide to use it, ie, when we place a warmup.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -129,6 +132,58 @@ private static boolean isRestoring(final MemberTaskOffsets
memberTaskOffsets, fi
return offsetOf(memberTaskOffsets.taskOffsets(), task) != null;
}
+ /**
+ * Indexes how loaded each process is, for the order in which the budget
pass funds warm-up tasks. The load of a
+ * process is its stateful task count over the number of members it runs
-- the same shape as the task assignor's
+ * own {@code ProcessState.load()}, so that both layers rank processes
comparably.
+ *
+ * <p><b>Only stateful tasks are counted</b>, which is narrower than what
the assignor measures. Standby and
+ * warm-up tasks exist only for stateful tasks anyway, so in practice this
comes down to leaving stateless active
+ * tasks out, for two reasons. Where the assignor spreads stateless tasks
evenly, they add the same amount to
+ * every process's load and so cannot change the ranking at all. Where it
does not spread them evenly, only
+ * stateful work competes for the changelog reads a warm-up needs, so
counting stateless tasks would rank a
+ * process busy with work that does not compete as though it were a poor
place to restore.
+ *
+ * <p>A process running nothing but stateless tasks therefore has a load
of zero, which is the right answer
+ * here. That it holds no state to take over is beside the point: the
target assignment has already chosen every
+ * destination, and this order only decides which of those migrations is
funded first, never where a task goes.
+ *
+ * <p>Only {@link StreamsGroupMember#assignedTasks()} is counted -- {@link
+ * StreamsGroupMember#tasksPendingRevocation()} is deliberately not read,
and the two are disjoint, so nothing on
+ * its way out is counted. Counting a task the member has been told to
give up would overstate the load the
+ * process is about to carry, and would double-count the commonest shape
of all: a member being demoted from
+ * active to standby holds the task as a pending active revocation and as
an already-granted standby at once.
+ *
+ * @param members
+ * All members of the group.
+ * @param subtopologies
+ * The resolved subtopologies, which tell whether a subtopology is
stateful.
+ *
+ * @return The load of every process running at least one member, indexed
by process ID.
+ */
+ static Map<String, ProcessLoad> indexProcessLoad(
+ final Map<String, StreamsGroupMember> members,
+ final SortedMap<String, ConfiguredSubtopology> subtopologies
+ ) {
+ final Map<String, Integer> memberCounts = new HashMap<>();
+ final Map<String, Integer> statefulTaskCounts = new HashMap<>();
+
+ for (final StreamsGroupMember member : members.values()) {
+ final String processId = member.processId();
+ memberCounts.merge(processId, 1, Integer::sum);
Review Comment:
Did not benchmark this yet, but fully agree, it's worth to check out... \cc
@suzhiking
Claude generate this, and I actually had the same thought, but left it as-if
for now.
##########
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
Review Comment:
It's the last bullet point (`A standby on a <b>sibling</b> member`), but
agreed, the reference is not clear.
##########
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
Review Comment:
It means we warmup state on the target owner before we re-assign the task
from old owner to the new owner (ie, the task is migrated from old owner to new
owner).
##########
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;
+ }
+
+ /**
+ * Whether the migration can still be warmed for free if it does not get a
slot, by leaving a standby the target
+ * owner's process holds on one of its <em>other</em> members where it is.
Such a standby goes on consuming from
+ * the changelog wherever it sits, so it warms the destination process
either way; what the slot buys is moving it
+ * onto the target owner, so that the hand-over becomes an in-place
promotion instead of a release and reopen.
+ *
+ * <p>A standby on the target owner itself is not covered here: that one
is borrowed outright and never competes
+ * for a slot, so it never reaches the point of needing a fallback.
+ */
+ private static boolean isBorrowableFromSibling(final StagedMigration
migration) {
Review Comment:
Ack.
##########
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()) {
Review Comment:
Can you define "high"? -- A heap has certain overhead by itself, and the
thinking was that it might not be necessary to use one?
Claude:
> The loop is O(B·n) with B = min(cap, n), the cap defaults to 2 with a
ceiling of 20, so 1000 candidates at B=20 is ~20k comparisons and zero
allocation. The bucketed-heap alternative pays O(n log n) up front to order
candidates it will mostly never examine — its cost scales with the candidate
count where the scan's scales with the budget — and its process comparator has
to carry the bucket head's secondary keys or cross-process ties break
inconsistently with the global order.
Let me know what you think. We can of course also experiment and benchmark
later. \cc @suzhiking
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -129,6 +132,58 @@ private static boolean isRestoring(final MemberTaskOffsets
memberTaskOffsets, fi
return offsetOf(memberTaskOffsets.taskOffsets(), task) != null;
}
+ /**
+ * Indexes how loaded each process is, for the order in which the budget
pass funds warm-up tasks. The load of a
+ * process is its stateful task count over the number of members it runs
-- the same shape as the task assignor's
+ * own {@code ProcessState.load()}, so that both layers rank processes
comparably.
+ *
+ * <p><b>Only stateful tasks are counted</b>, which is narrower than what
the assignor measures. Standby and
+ * warm-up tasks exist only for stateful tasks anyway, so in practice this
comes down to leaving stateless active
+ * tasks out, for two reasons. Where the assignor spreads stateless tasks
evenly, they add the same amount to
+ * every process's load and so cannot change the ranking at all. Where it
does not spread them evenly, only
+ * stateful work competes for the changelog reads a warm-up needs, so
counting stateless tasks would rank a
+ * process busy with work that does not compete as though it were a poor
place to restore.
+ *
+ * <p>A process running nothing but stateless tasks therefore has a load
of zero, which is the right answer
+ * here. That it holds no state to take over is beside the point: the
target assignment has already chosen every
+ * destination, and this order only decides which of those migrations is
funded first, never where a task goes.
+ *
+ * <p>Only {@link StreamsGroupMember#assignedTasks()} is counted -- {@link
+ * StreamsGroupMember#tasksPendingRevocation()} is deliberately not read,
and the two are disjoint, so nothing on
+ * its way out is counted. Counting a task the member has been told to
give up would overstate the load the
+ * process is about to carry, and would double-count the commonest shape
of all: a member being demoted from
+ * active to standby holds the task as a pending active revocation and as
an already-granted standby at once.
+ *
+ * @param members
+ * All members of the group.
+ * @param subtopologies
+ * The resolved subtopologies, which tell whether a subtopology is
stateful.
+ *
+ * @return The load of every process running at least one member, indexed
by process ID.
+ */
+ static Map<String, ProcessLoad> indexProcessLoad(
+ final Map<String, StreamsGroupMember> members,
+ final SortedMap<String, ConfiguredSubtopology> subtopologies
+ ) {
+ final Map<String, Integer> memberCounts = new HashMap<>();
+ final Map<String, Integer> statefulTaskCounts = new HashMap<>();
+
+ for (final StreamsGroupMember member : members.values()) {
+ final String processId = member.processId();
+ memberCounts.merge(processId, 1, Integer::sum);
+
+ final Consumer<TaskId> count = task ->
statefulTaskCounts.merge(processId, 1, Integer::sum);
+
forEachStatefulActiveTask(member.assignedTasks().activeTasksWithEpochs(),
subtopologies, count);
+ forEachStatefulTask(member.assignedTasks().standbyTasks(),
subtopologies, count);
+ forEachStatefulTask(member.assignedTasks().warmupTasks(),
subtopologies, count);
Review Comment:
> Are we sure we want to count warmupTasks
I believe yes. -- Question back: why not? A warmup is running in the same
sense as a standby is running, and it puts load on the member/process. -- Note,
it's possible that `num.warmup.replicas` config changes, and for this case, we
might be able to assign a new warmup, and want to put it on a member load-based
(and consider already assigned warmups. Also true for the other way around when
we need to revoke a warmup because out budged was decrease: we would revoke
from the highest loaded process/member.
> If I have already created the warmupTask in the last round, will I
reconsider it here?
No. The algorithm is design to avoid a "flip-flop" (maybe I should open a
stacked PR with the overall code). If we decided to put a warmup on a member,
we keep it there until it's hot in every refiner re-run. We would only change
the decision if a new rebalance happened and we get a new target assignment.
##########
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
Review Comment:
That's `num.warmup.replicas` -- let me clean this up
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -268,12 +323,24 @@ static TaskDecisions analyzeTasks(
* is dropped when the task closes, and no hand-over of a running task
between threads of one process exists to
* carry it across. Worse, the lag that made the task look ready was
measured on the member that is about to
* close, so for an in-memory store it says nothing about what the
incoming member then has to restore. This
- * predicate cannot fix that; it would take a client-side cross-thread
task hand-over. The broker cannot even see
- * the difference, because the topology metadata carries changelog topics
but not how a store is backed.
+ * predicate cannot fix that; it would take a client-side cross-thread
task hand-over
+ * (https://issues.apache.org/jira/browse/KAFKA-21090). The broker cannot
even see the difference, because the
+ * topology metadata carries changelog topics but not how a store is
backed.
+ *
+ * <p>The damage is bounded, though, because <b>the refiner never creates
one of those two paths -- it only ever
+ * inherits them.</b> Every warm-up it plants sits on the target owner
itself, so every migration it warms ends in
+ * the in-place promotion, which is warm for every store type. It even
pays to keep that true: where the
+ * destination process holds a copy of the task only on a <em>sibling</em>
of the target owner, the budget pass
+ * spends a slot to move that copy onto the target owner rather than
borrow it where it sits. So the only way to
+ * reach one of the two cold paths is through this predicate granting the
task outright -- nothing was warmed, and
+ * the layout was already there when the refiner looked.
*
- * <p>What bounds the damage is that a warm-up task the refiner plants
always targets the target owner itself, so
- * every migration the refiner stages resolves through the in-place
promotion. The other paths arise only out of a
- * layout the refiner inherited.
+ * <p>Note that includes a copy on a sibling member that is
<em>already</em> caught up: the task is granted here,
+ * in this step, before the budget pass ever sees the migration, so
nothing gets the chance to move that copy onto
+ * the target owner first. Doing so would spend a slot to buy an in-place
promotion -- which is worth it for an
+ * in-memory store and pure waste for a store that persists to disk, since
that one reopens warm from the state
+ * directory anyway. The broker cannot tell the two apart, so this grants
immediately and converges fast. It is a
+ * deliberate boundary rather than an oversight, and the design document
carries it as an open question.
Review Comment:
`isReady` decided if we are ready to grant/promote/assign and active task to
it's new owner. There is many different cases to consider, but they all fold
into a single predicate from below, so I try to cover the different cases in
the JavaDocs, because it's impossible to infer from the code.
Which part is particularly unclear (or which cases are clear)?
##########
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()));
Review Comment:
The difference is, that we use a custom `Cooperator` which takes the mutable
`processLoad` as parameter. While `processLoad` is not modified while we build
up the collection, but when we later read the collection, we update
`processLoad` as side effect breaking the search.
##########
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:
Only for completeness (to track all task movements we need to make it a full
set `warmupTasks u borrowedMigrations u parkedMigration`), and observability: I
was thinking to add a broker side metric (in a follow up PR) to be able to
monitor the end-to-end progress to reach the target assignment.
It's not functional, and we could also compute it (for metrics) from others
data structures, but it seemed convenient to track it explicitly.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -547,4 +897,83 @@ record TaskDecisions(
List<TaskGrant> grantedTasks
) {
}
+
+ /**
+ * What a staged migration needs from the budget, which is what the
classification pass sorts them by.
+ */
+ private enum Warming {
+ /** Nothing can warm this migration and no slot may be spent on it. */
+ PARK,
+
+ /** A standby on the target owner itself already warms it, for free. */
+ BORROW,
+
+ /** A warm-up is already restoring for it, and keeps the slot it was
funded with. */
+ KEEP,
+
+ /** It needs a warm-up placed on its target owner, which costs a slot.
*/
+ PLANT
+ }
+
+ /**
+ * A staged migration competing for a warm-up slot, with the parts of the
funding order that can be resolved
+ * ahead of the comparisons.
+ *
+ * @param task
+ * The task being migrated.
+ * @param targetOwner
+ * The member the warm-up task goes on, if this migration is
funded. Always the migration's target
+ * owner, so that the warm-up can be promoted in place once it has
caught up.
+ * @param targetProcessId
+ * The process that member runs in, whose load the funding order
reads and the accounting raises.
+ * @param sourceLoad
+ * The load of the process still running the task, which cannot
change during a funding pass.
+ * @param borrowable
+ * Whether missing out on a slot leaves the migration warmed
anyway, because a standby on a sibling member
+ * of the target owner's process can be borrowed where it sits.
Such a candidate never parks.
+ */
+ private record FundingCandidate(
+ TaskId task,
+ String targetOwner,
+ String targetProcessId,
+ double sourceLoad,
Review Comment:
We fund a warmup on the _target_, not the source. The source is the old
owner, and if we put a warmup on the target, and keep the active task on the
source until the warmup is hot, so the source load does not change.
##########
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 fro 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]
