lucasbru commented on code in PR #23484:
URL: https://github.com/apache/kafka/pull/23484#discussion_r4035176279
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -520,6 +521,204 @@ private static double currentProcessLoad(
.loadWith(newWarmupsByProcess.getOrDefault(candidate.currentProcessId(), 0));
}
+ /**
+ * Decides which of the target assignment's standby placements this step
has to hold back.
+ *
+ * <p>Nothing is invented or dropped permanently: every placement comes
from the target assignment, and one held
+ * back here is emitted by a later step once its reason is gone. A
placement of task {@code t} on member {@code m}
+ * of process {@code p} is withheld when:
+ * <ol>
+ * <li>{@code p} currently runs {@code t} as an active task, because
{@code t}'s migration off {@code p} is
+ * staged: F moved {@code t}'s active off {@code p} and relocated its
standby, but the refiner held the active
Review Comment:
what's F? I'm not sure what the second part of this paragraph adds / means.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -520,6 +521,204 @@ private static double currentProcessLoad(
.loadWith(newWarmupsByProcess.getOrDefault(candidate.currentProcessId(), 0));
}
+ /**
+ * Decides which of the target assignment's standby placements this step
has to hold back.
+ *
+ * <p>Nothing is invented or dropped permanently: every placement comes
from the target assignment, and one held
+ * back here is emitted by a later step once its reason is gone. A
placement of task {@code t} on member {@code m}
+ * of process {@code p} is withheld when:
+ * <ol>
+ * <li>{@code p} currently runs {@code t} as an active task, because
{@code t}'s migration off {@code p} is
+ * staged: F moved {@code t}'s active off {@code p} and relocated its
standby, but the refiner held the active
+ * in place, so F's standby cannot land on {@code p} yet.</li>
+ * <li>{@code t}'s migration onto {@code p} borrowed an existing
standby on {@code p}. To not run
+ * {@code num.standby.repliacs + 1} standbys, we hold back the
assignment of the standby to its new owner.</li>
+ * </ol>
+ *
+ * @param targetAssignment
+ * All members' target assignments, as computed by the task
assignor.
+ * @param currentAssignment
+ * The indexed current assignment, from {@link
#indexCurrentAssignment}.
+ * @param decisions
+ * What the case analysis decided, from {@link #analyzeTasks}.
+ * @param warmupPlan
+ * How each staged migration is being warmed, from {@link
#planWarmups}.
+ * @param members
+ * All members of the group, used to resolve which process a member
runs in.
+ * @param subtopologies
+ * The resolved subtopologies, which tell whether a subtopology is
stateful.
+ *
+ * @return The standby placements to withhold, as the tasks to drop from
each member's slice, in canonical order.
+ * A member with nothing withheld does not appear.
+ */
+ static SortedMap<String, SortedSet<TaskId>> filterStandbys(
+ final Map<String, TasksTuple> targetAssignment,
+ final CurrentAssignmentIndex currentAssignment,
+ final TaskDecisions decisions,
+ final WarmupPlan warmupPlan,
+ final Map<String, StreamsGroupMember> members,
+ final SortedMap<String, ConfiguredSubtopology> subtopologies
+ ) {
+ final StandbyConflicts conflicts =
+ indexStandbyConflicts(currentAssignment, decisions, warmupPlan,
members);
+ final SortedMap<String, SortedSet<TaskId>> withheld = new TreeMap<>();
+
+ targetAssignment.forEach((memberId, tasks) -> {
+ final StreamsGroupMember member = members.get(memberId);
+ if (member == null) {
+ // The target assignment can name a member the group has
already removed. Its slice reaches nobody, so
+ // there is nothing to hold back and no process to resolve it
against.
+ return;
+ }
+
+ forEachStatefulTask(tasks.standbyTasks(), subtopologies, task -> {
+ if (isStandbyWithheld(memberId, member.processId(), task,
currentAssignment, conflicts)) {
+ withheld.computeIfAbsent(memberId, __ -> new
TreeSet<>()).add(task);
+ }
+ });
+ });
+
+ return Collections.unmodifiableSortedMap(withheld);
+ }
+
+ /**
+ * Builds the {@link StandbyConflicts} lookups, ie, where each task is
held active, which tasks are granted
+ * this step, and which migrations borrowed a copy.
+ */
+ private static StandbyConflicts indexStandbyConflicts(
+ final CurrentAssignmentIndex currentAssignment,
+ final TaskDecisions decisions,
+ final WarmupPlan warmupPlan,
+ final Map<String, StreamsGroupMember> members
+ ) {
+ final Map<TaskId, String> activeHeldOn = new HashMap<>();
+ currentAssignment.activeHolder().forEach((task, holder) ->
+ activeHeldOn.put(task,
members.get(holder.memberId()).processId()));
+
+ final Set<TaskId> grantedTasks = new HashSet<>();
+ for (final TaskGrant grant : decisions.grantedTasks()) {
+ grantedTasks.add(grant.task());
+ }
+
+ return new StandbyConflicts(activeHeldOn, grantedTasks,
warmupPlan.borrowedMigrations());
+ }
+
+ /**
+ * Whether this step has to hold the standby placement back.
+ */
+ private static boolean isStandbyWithheld(
+ final String memberId,
+ final String processId,
+ final TaskId task,
+ final CurrentAssignmentIndex currentAssignment,
+ final StandbyConflicts conflicts
+ ) {
+ final ActiveHolder activeHolder =
currentAssignment.activeHolder().get(task);
+
+ // Rule 1: a process cannot hold `task` twice, so if this standby's
process already runs the active,
+ // withhold the standby -- unless it is the demotion (this member is
handing the active away this step and
+ // relabels it into the standby in place).
Review Comment:
Not sure if the term "demotion" is helpful here. Aren't we just checking if
the active task is migrated away in this step?
What if the active holder is a sibling of the `memberId`, but the task
migration is granted -- then I don't have to withhold right?
What if the active holder is `memberId`, but the task migration that was
granted is into a sibling -- then I do have to withhold right?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -520,6 +521,204 @@ private static double currentProcessLoad(
.loadWith(newWarmupsByProcess.getOrDefault(candidate.currentProcessId(), 0));
}
+ /**
+ * Decides which of the target assignment's standby placements this step
has to hold back.
+ *
+ * <p>Nothing is invented or dropped permanently: every placement comes
from the target assignment, and one held
+ * back here is emitted by a later step once its reason is gone. A
placement of task {@code t} on member {@code m}
+ * of process {@code p} is withheld when:
+ * <ol>
+ * <li>{@code p} currently runs {@code t} as an active task, because
{@code t}'s migration off {@code p} is
+ * staged: F moved {@code t}'s active off {@code p} and relocated its
standby, but the refiner held the active
+ * in place, so F's standby cannot land on {@code p} yet.</li>
+ * <li>{@code t}'s migration onto {@code p} borrowed an existing
standby on {@code p}. To not run
+ * {@code num.standby.repliacs + 1} standbys, we hold back the
assignment of the standby to its new owner.</li>
Review Comment:
nit: replicas
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -520,6 +521,204 @@ private static double currentProcessLoad(
.loadWith(newWarmupsByProcess.getOrDefault(candidate.currentProcessId(), 0));
}
+ /**
+ * Decides which of the target assignment's standby placements this step
has to hold back.
+ *
+ * <p>Nothing is invented or dropped permanently: every placement comes
from the target assignment, and one held
+ * back here is emitted by a later step once its reason is gone. A
placement of task {@code t} on member {@code m}
+ * of process {@code p} is withheld when:
+ * <ol>
+ * <li>{@code p} currently runs {@code t} as an active task, because
{@code t}'s migration off {@code p} is
+ * staged: F moved {@code t}'s active off {@code p} and relocated its
standby, but the refiner held the active
+ * in place, so F's standby cannot land on {@code p} yet.</li>
+ * <li>{@code t}'s migration onto {@code p} borrowed an existing
standby on {@code p}. To not run
+ * {@code num.standby.repliacs + 1} standbys, we hold back the
assignment of the standby to its new owner.</li>
+ * </ol>
+ *
+ * @param targetAssignment
+ * All members' target assignments, as computed by the task
assignor.
+ * @param currentAssignment
+ * The indexed current assignment, from {@link
#indexCurrentAssignment}.
+ * @param decisions
+ * What the case analysis decided, from {@link #analyzeTasks}.
+ * @param warmupPlan
+ * How each staged migration is being warmed, from {@link
#planWarmups}.
+ * @param members
+ * All members of the group, used to resolve which process a member
runs in.
+ * @param subtopologies
+ * The resolved subtopologies, which tell whether a subtopology is
stateful.
+ *
+ * @return The standby placements to withhold, as the tasks to drop from
each member's slice, in canonical order.
Review Comment:
What is a member's slice.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -520,6 +521,204 @@ private static double currentProcessLoad(
.loadWith(newWarmupsByProcess.getOrDefault(candidate.currentProcessId(), 0));
}
+ /**
+ * Decides which of the target assignment's standby placements this step
has to hold back.
+ *
+ * <p>Nothing is invented or dropped permanently: every placement comes
from the target assignment, and one held
+ * back here is emitted by a later step once its reason is gone. A
placement of task {@code t} on member {@code m}
+ * of process {@code p} is withheld when:
+ * <ol>
+ * <li>{@code p} currently runs {@code t} as an active task, because
{@code t}'s migration off {@code p} is
+ * staged: F moved {@code t}'s active off {@code p} and relocated its
standby, but the refiner held the active
+ * in place, so F's standby cannot land on {@code p} yet.</li>
+ * <li>{@code t}'s migration onto {@code p} borrowed an existing
standby on {@code p}. To not run
Review Comment:
Remind me again we we cannot just turn the standby into a warm-up task. Is
the point so that we do not have to spend a slot?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -520,6 +521,204 @@ private static double currentProcessLoad(
.loadWith(newWarmupsByProcess.getOrDefault(candidate.currentProcessId(), 0));
}
+ /**
+ * Decides which of the target assignment's standby placements this step
has to hold back.
+ *
+ * <p>Nothing is invented or dropped permanently: every placement comes
from the target assignment, and one held
+ * back here is emitted by a later step once its reason is gone. A
placement of task {@code t} on member {@code m}
+ * of process {@code p} is withheld when:
+ * <ol>
+ * <li>{@code p} currently runs {@code t} as an active task, because
{@code t}'s migration off {@code p} is
+ * staged: F moved {@code t}'s active off {@code p} and relocated its
standby, but the refiner held the active
+ * in place, so F's standby cannot land on {@code p} yet.</li>
+ * <li>{@code t}'s migration onto {@code p} borrowed an existing
standby on {@code p}. To not run
+ * {@code num.standby.repliacs + 1} standbys, we hold back the
assignment of the standby to its new owner.</li>
+ * </ol>
+ *
+ * @param targetAssignment
+ * All members' target assignments, as computed by the task
assignor.
+ * @param currentAssignment
+ * The indexed current assignment, from {@link
#indexCurrentAssignment}.
+ * @param decisions
+ * What the case analysis decided, from {@link #analyzeTasks}.
+ * @param warmupPlan
+ * How each staged migration is being warmed, from {@link
#planWarmups}.
+ * @param members
+ * All members of the group, used to resolve which process a member
runs in.
+ * @param subtopologies
+ * The resolved subtopologies, which tell whether a subtopology is
stateful.
+ *
+ * @return The standby placements to withhold, as the tasks to drop from
each member's slice, in canonical order.
+ * A member with nothing withheld does not appear.
+ */
+ static SortedMap<String, SortedSet<TaskId>> filterStandbys(
+ final Map<String, TasksTuple> targetAssignment,
+ final CurrentAssignmentIndex currentAssignment,
+ final TaskDecisions decisions,
+ final WarmupPlan warmupPlan,
+ final Map<String, StreamsGroupMember> members,
+ final SortedMap<String, ConfiguredSubtopology> subtopologies
+ ) {
+ final StandbyConflicts conflicts =
+ indexStandbyConflicts(currentAssignment, decisions, warmupPlan,
members);
+ final SortedMap<String, SortedSet<TaskId>> withheld = new TreeMap<>();
+
+ targetAssignment.forEach((memberId, tasks) -> {
+ final StreamsGroupMember member = members.get(memberId);
+ if (member == null) {
+ // The target assignment can name a member the group has
already removed. Its slice reaches nobody, so
+ // there is nothing to hold back and no process to resolve it
against.
+ return;
+ }
+
+ forEachStatefulTask(tasks.standbyTasks(), subtopologies, task -> {
+ if (isStandbyWithheld(memberId, member.processId(), task,
currentAssignment, conflicts)) {
+ withheld.computeIfAbsent(memberId, __ -> new
TreeSet<>()).add(task);
+ }
+ });
+ });
+
+ return Collections.unmodifiableSortedMap(withheld);
+ }
+
+ /**
+ * Builds the {@link StandbyConflicts} lookups, ie, where each task is
held active, which tasks are granted
+ * this step, and which migrations borrowed a copy.
+ */
+ private static StandbyConflicts indexStandbyConflicts(
+ final CurrentAssignmentIndex currentAssignment,
+ final TaskDecisions decisions,
+ final WarmupPlan warmupPlan,
+ final Map<String, StreamsGroupMember> members
+ ) {
+ final Map<TaskId, String> activeHeldOn = new HashMap<>();
+ currentAssignment.activeHolder().forEach((task, holder) ->
+ activeHeldOn.put(task,
members.get(holder.memberId()).processId()));
+
+ final Set<TaskId> grantedTasks = new HashSet<>();
+ for (final TaskGrant grant : decisions.grantedTasks()) {
+ grantedTasks.add(grant.task());
+ }
+
+ return new StandbyConflicts(activeHeldOn, grantedTasks,
warmupPlan.borrowedMigrations());
+ }
+
+ /**
+ * Whether this step has to hold the standby placement back.
+ */
+ private static boolean isStandbyWithheld(
+ final String memberId,
+ final String processId,
+ final TaskId task,
+ final CurrentAssignmentIndex currentAssignment,
+ final StandbyConflicts conflicts
+ ) {
+ final ActiveHolder activeHolder =
currentAssignment.activeHolder().get(task);
+
+ // Rule 1: a process cannot hold `task` twice, so if this standby's
process already runs the active,
+ // withhold the standby -- unless it is the demotion (this member is
handing the active away this step and
+ // relabels it into the standby in place).
+ if (processId.equals(conflicts.activeHeldOn().get(task))) {
+ final boolean isDemotion =
+ memberId.equals(activeHolder.memberId()) &&
conflicts.grantedTasks().contains(task);
+ return !isDemotion;
+ }
+
+ // Rule 2: a borrowed migration keeps its existing copy as the group's
one entitled replica, so F's
Review Comment:
another mention of F
--
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]
