lucasbru commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4005744625
##########
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:
When you benchmark this, you will notice that these kinds of higher-level
operations are much slower than a for loop.
##########
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
Review Comment:
very wordy indeed.
"We only count stateful tasks. This is not used to decide the assignment,
just to decide in which order task movements are funded." would possibly be
enough.
##########
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:
What is a standing cap.
##########
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 haven't grasped the full
algorithm yet, just leaving a note due to the possibly circular nature of
reasoning -- If I have already created the warmupTask in the last round, will
I reconsider it here?
##########
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:
For borrowedMigration you use a sorted Set, here you use a list + sort. Why?
##########
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:
What is "that move" referring to?
##########
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:
What does "warming the migration" mean?
##########
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:
for high max.warmup.replicas, this loop can explode. Considered using a
Priority Heap and pop off in log(n)
##########
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:
Why does the source load not change during the funding - if I fund warm ups
on the source, the load should go up?
##########
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:
Seems like we already check this predicate in `warmingOf`. Maybe worth
keeping a separate category?
##########
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:
Why do we need to know about parked migrations?
##########
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.
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?
##########
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:
I have read this javadoc three times but I'm still not sure I get it. What
are slots, how do you pay for 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]
