ChoMinGi commented on code in PR #23454:
URL: https://github.com/apache/kafka/pull/23454#discussion_r4033906858
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -547,4 +805,92 @@ record TaskDecisions(
List<TaskGrant> grantedTasks
) {
}
+
+ /**
+ * What a staged migration needs from the warm-up budget. This is decided
entirely by what the target owner's
+ * process already holds for the task, so the five values are mutually
exclusive and cover every staged
+ * migration.
+ */
+ private enum Warming {
+ /** Nothing can warm this migration and no warm-up 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 warm-up slot
it was funded with. */
+ KEEP,
+
+ /** Its target owner's process holds nothing for the task: a warm-up
must be planted. Costs a warm-up slot. */
+ PLANT,
+
+ /**
+ * Its target owner's process holds a standby of the task, but on one
of its <em>other</em> members.
+ * Cost a warmu-up slot.
+ */
+ SIBLING_MOVE
+ }
+
+ /**
+ * 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 currentProcessId
+ * The process still running the task, whose load the funding order
reads (descending) as its secondary key.
+ * @param warming
+ * What this migration needs from the budget, decided once when the
migration is classified. Only the
+ * fall-back turns on it: a {@link Warming#SIBLING_MOVE} that does
not get a warm-up slot falls back to
+ * borrowing the standby where it sits, everything else parks.
+ */
+ private record FundingCandidate(
+ TaskId task,
+ String targetOwner,
+ String targetProcessId,
+ String currentProcessId,
+ Warming warming
+ ) {
+ }
+
+ /**
+ * Which warm-up tasks the intermediate assignment places, and how each
staged migration is being warmed.
+ *
+ * <p>Every migration the case analysis staged appears in exactly one of
these: its task is either a key of
+ * {@code warmupTasks}, or in {@code borrowedMigrations}, or in {@code
parkedMigrations}.
+ *
+ * @param warmupTasks
+ * The member holding a warm-up task of each task, which is always
that task's target owner. A warm-up
+ * task planted in this step and one that has been restoring for
several look alike here; nothing
+ * downstream needs the difference, and {@link
CurrentAssignmentIndex#taskCopies()} still tells them
+ * apart.
+ * @param borrowedMigrations
+ * The migrations warmed for free by a standby the target owner's
process already holds, which it keeps where
+ * it is. They spend no warm-up slot, on the condition that the
standby filter withholds the copy the
+ * target assignment relocates in its place. Either the standby
sits on the target owner itself, where it
+ * is borrowed outright because the promotion can take it over in
place, or it sits on a sibling member and
+ * the migration competed for a warm-up slot to move it across, did
not get one, and settles for warming
+ * through the sibling.
+ * @param parkedMigrations
+ * The migrations with nothing warming them: the budget was spent,
or the target assignment names a
+ * member the group no longer has. Their tasks keep running on
their current owners (or temporarily not at
+ * all), and a later refinement step picks them up.
+ */
+ record WarmupPlan(
+ SortedMap<TaskId, String> warmupTasks,
+ SortedSet<TaskId> borrowedMigrations,
+ SortedSet<TaskId> parkedMigrations
+ ) {
+
+ static final WarmupPlan EMPTY = new WarmupPlan(
+ Collections.emptySortedMap(),
+ Collections.emptySortedSet(),
+ Collections.emptySortedSet()
+ );
+ }
}
Review Comment:
For `SIBLING_MOVE`, indexProcessLoad() already counts the standby on the
target process. Since the warm-up replaces it on the same process, shouldn't
the process load stay unchanged?
Should this increment apply only to `PLANT`?
--
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]
