lucasbru commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4024835300
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -202,20 +232,15 @@ static TaskDecisions analyzeTasks(
Optional.empty()
));
}
- // A task nobody holds and whose target owner is gone is left
to the next assignor run: granting it to
- // a member that is no longer in the group would achieve
nothing.
continue;
}
final String targetProcessId = targetMember.processId();
- // The task moves now, for any of three reasons. Nobody holds it
-- it is new, its owner left, or a
- // hand-over is in flight and the previous owner has already
released it -- so there is nothing to protect
- // and the target owner takes it even cold; preferring a warmer
owner would be a placement decision, and
- // placement is the assignor's job. Or somebody holds it but is
still restoring it, so nothing is being
- // processed and staging would protect nothing. Or somebody is
processing it but no achievable warming
- // improvement remains. The holder, when there is one, is
necessarily a member of the group, because the
- // index it comes from was built from the members themselves.
+ // The task moves now, for any of three reasons.
+ // 1. Nobody holds it.
+ // 2. Somebody holds it but is still restoring it.
+ // 3. Somebody is processing it and the target owener is caught
up
Review Comment:
nit: owener -> owner
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -236,44 +261,18 @@ static TaskDecisions analyzeTasks(
}
/**
- * Whether no achievable warming improvement remains for handing the task
over to its target owner.
+ * Whether a task is ready for migration to its target owner.
*
- * <p>This is deliberately weaker than "the target owner has caught up".
Warming up is only worth staging when it
- * can actually shorten the hand-over, and there are two situations where
it cannot -- one per clause of the
- * predicate:
- * <ul>
- * <li><b>The task is moving between two members of one process.</b> A
process must not hold the same task
- * twice, so there is no way to warm the target owner up while the
current owner still runs it. No condition
- * on the state applies here: staging such a move would park it
forever, so it has to count as ready.</li>
- * <li><b>The target owner's process already holds a caught-up copy of
the task.</b> Here <em>caught up</em>
- * carries the weight: a copy that is still catching up leaves a
genuine improvement to wait for, so the task
- * stays staged, that copy keeps consuming, and a later step grants
the task once the copy is hot. Planting a
- * warm-up task on the target owner is no help either way, because its
process would then hold the task
- * twice. The copy sits either on the target owner itself, which
promotes it in place, or on a sibling
- * member, which has to release the task first so that the target
owner can reopen it from the state
- * directory.</li>
- * </ul>
+ * <p>The general case is, the warmup is caught up on the target
<b>member</b>, but there are two more cases we need
+ * to consider:
+ * <b>(1) The task is moving between two members of one process:</b> A
process cannot hold two copies of a task at
+ * the same time, so we cannot put a warmup but can only migrate the task
right away.
+ * <b>(2) The task is moved to a different process, but a sibling member
of the target member holds a copy of the
+ * task:</b> Similar to case (1), a process cannot hold two copies of the
same task at the same time; however, in
+ * contrast to (1) we can wait until the sibling copy gets caught up
before we do the migration.
Review Comment:
"we can wait until the sibling copy gets caught up before we do the
migration".
But this is not what happens right? According to the javadoc in planWarmups,
we don't wait for the sibling copy to get caught up, we move the task to the
target member.
The code is right - we do not consider it ready in this case. But the
javadoc contradicts planWarmups.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -288,6 +287,226 @@ private static boolean isReady(
.anyMatch(holder -> holder.processId().equals(targetProcessId) &&
holder.caughtUp());
}
+ /**
+ * Decides which of the staged migrations get a warm-up task, under the
warmup budget.
+ *
+ * <p>There is different scenarios:
+ * <ul>
+ * <li>A warm-up task already restoring keeps its warm-up slot if the
target assignment didn't change, and the
+ * warmu-up task is not caught up yet. It could also get revoked if
the warmup budget was reduced and keeping
+ * the warmup would now exceed the budget.
+ * <li>A <b>fresh plant</b> puts a warm-up task on a target owner
whose process holds nothing for the task,
+ * and spends a warm-up slot.</li>
+ * <li>When the target owner <em>itself</em> already holds a standby
of the task we can <b>borrow</b> it,
+ * and no warmup budget is used: that standby warms-up the task
anyway.</li>
Review Comment:
So this borrow case excludes standbys that are caught-up, because these
would be granted right away, right?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -288,6 +287,226 @@ private static boolean isReady(
.anyMatch(holder -> holder.processId().equals(targetProcessId) &&
holder.caughtUp());
}
+ /**
+ * Decides which of the staged migrations get a warm-up task, under the
warmup budget.
+ *
+ * <p>There is different scenarios:
+ * <ul>
+ * <li>A warm-up task already restoring keeps its warm-up slot if the
target assignment didn't change, and the
+ * warmu-up task is not caught up yet. It could also get revoked if
the warmup budget was reduced and keeping
+ * the warmup would now exceed the budget.
+ * <li>A <b>fresh plant</b> puts a warm-up task on a target owner
whose process holds nothing for the task,
+ * and spends a warm-up slot.</li>
+ * <li>When the target owner <em>itself</em> already holds a standby
of the task we can <b>borrow</b> it,
+ * and no warmup budget is used: that standby warms-up the task
anyway.</li>
+ * <li>If a target member's <em>sibling</em> hold a standby, we cannot
borrow but, but need to move the
+ * standby to its new owner, and putting a warmup on the target
member, spending a warm-up slot.
+ * (Cf case (2) of {@link #isReady(CurrentAssignmentIndex, TaskId,
String, String)} </li>
+ * </ul>
+ *
+ * <p>Everything else <b>parks</b> -- the task keeps running on its
current owner with nothing warming up, and a
+ * later refinement step picks it up once a warm-up slot frees.
+ *
+ * @param decisions
+ * What the case analysis decided, from {@link #analyzeTasks}.
+ * @param members
+ * All members of the group, used to resolve which process a task's
current owner runs in.
+ * @param processLoad
+ * The load of each process, from {@link #indexProcessLoad}.
+ * @param numWarmupReplicas
+ * How many copies beyond the target assignment may exist at once,
group-wide.
+ *
+ * @return Which warm-up tasks the intermediate assignment places, and how
each staged migration is warmed.
+ */
+ static WarmupPlan planWarmups(
+ final TaskDecisions decisions,
+ final Map<String, StreamsGroupMember> members,
+ final Map<String, ProcessLoad> processLoad,
+ final int numWarmupReplicas
+ ) {
+ if (numWarmupReplicas == 0) {
+ return WarmupPlan.EMPTY;
+ }
+
+ final SortedMap<TaskId, String> warmupTasks = new TreeMap<>();
+ final SortedSet<TaskId> borrowedMigrations = new TreeSet<>();
+ final SortedSet<TaskId> parkedMigrations = new TreeSet<>();
+
+ final List<FundingCandidate> keptWarmups = new ArrayList<>();
+ final List<FundingCandidate> newWarmupCandidates = new ArrayList<>();
+
+ for (final StagedMigration migration : decisions.stagedMigrations()) {
+ final Warming warming = warmingOf(migration);
+ switch (warming) {
+ case PARK -> parkedMigrations.add(migration.task());
+ case BORROW -> borrowedMigrations.add(migration.task());
+ case KEEP -> keptWarmups.add(fundingCandidate(migration,
members, processLoad, warming));
+ // Both cases put a warm-up task on the target owner and both
cost a warm-up slot, so they share one
+ // candidate list and compete on equal terms.
+ case PLANT, SIBLING_MOVE ->
+ newWarmupCandidates.add(fundingCandidate(migration,
members, processLoad, warming));
+ }
+ }
+
+ // Warm-up tasks already restoring are funded first. If
`max.warmup.replicas` config was reduced, we might
+ // be over warmup budget and have to give up some warmup tasks.
Evicting in reverse funding order
+ // keeps which ones deterministic rather than dependent on iteration
order.
+ // Note: revocation of warmup task happens automatically by not adding
them to the assignment patch again
+ keptWarmups.sort((left, right) -> comparePriority(left, right,
processLoad, Map.of()));
+ for (int i = 0; i < keptWarmups.size(); i++) {
+ final FundingCandidate keptWarmup = keptWarmups.get(i);
+ if (i < numWarmupReplicas) {
+ warmupTasks.put(keptWarmup.task(), keptWarmup.targetOwner());
+ } else {
+ parkedMigrations.add(keptWarmup.task());
+ }
+ }
+
+ // New warm-up tasks raises its target process's load, so we need to
update it while we go, and find a new
+ // `best` from scratch each time
+ // note: this nested-loop is bounded by the number of unsed warm-up
slots; so while it's O(unsed * candidate)
+ // it's effectively not quadratic (we can consider `unsed` a constant)
Review Comment:
nit: unsed -> unused.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -288,6 +287,226 @@ private static boolean isReady(
.anyMatch(holder -> holder.processId().equals(targetProcessId) &&
holder.caughtUp());
}
+ /**
+ * Decides which of the staged migrations get a warm-up task, under the
warmup budget.
+ *
+ * <p>There is different scenarios:
+ * <ul>
+ * <li>A warm-up task already restoring keeps its warm-up slot if the
target assignment didn't change, and the
+ * warmu-up task is not caught up yet. It could also get revoked if
the warmup budget was reduced and keeping
+ * the warmup would now exceed the budget.
+ * <li>A <b>fresh plant</b> puts a warm-up task on a target owner
whose process holds nothing for the task,
+ * and spends a warm-up slot.</li>
+ * <li>When the target owner <em>itself</em> already holds a standby
of the task we can <b>borrow</b> it,
+ * and no warmup budget is used: that standby warms-up the task
anyway.</li>
+ * <li>If a target member's <em>sibling</em> hold a standby, we cannot
borrow but, but need to move the
+ * standby to its new owner, and putting a warmup on the target
member, spending a warm-up slot.
+ * (Cf case (2) of {@link #isReady(CurrentAssignmentIndex, TaskId,
String, String)} </li>
Review Comment:
cf case (2) of isReady - javadocs contract, I suppose the code is doing the
right thing: We move the copy from the sibling to the target member.
--
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]
