Copilot commented on code in PR #23484:
URL: https://github.com/apache/kafka/pull/23484#discussion_r4043892789
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -521,6 +522,208 @@ 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>a staged migration keeps {@code t} running as an active task on
{@code p}, and a process cannot hold
+ * {@code t} twice.</li>
+ * <li>{@code t}'s migration onto {@code p} borrowed an existing
standby on {@code p}. To not run
+ * {@code num.standby.replicas + 1} standbys, we hold back one
assignment of the standby to a 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 target assignment, 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(
+ targetAssignment,
+ currentAssignment,
+ decisions,
+ warmupPlan,
+ members,
+ subtopologies
+ );
+ 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 tasks reach 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,
conflicts)) {
+ withheld.computeIfAbsent(memberId, __ -> new
TreeSet<>()).add(task);
+ }
+ });
+ });
+
+ return Collections.unmodifiableSortedMap(withheld);
+ }
+
+ /**
+ * Builds the {@link StandbyConflicts} lookups, ie, where each staged
migration keeps its task running, and which
+ * placement pays for each borrowed copy.
+ */
+ private static StandbyConflicts indexStandbyConflicts(
+ 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 Map<TaskId, String> activeStagedOn = new HashMap<>();
+ for (final StagedMigration migration : decisions.stagedMigrations()) {
+ activeStagedOn.put(migration.task(),
members.get(migration.currentOwner()).processId());
+ }
+
+ final Map<TaskId, String> borrowPaidBy = new HashMap<>();
+ targetAssignment.forEach((memberId, tasks) ->
+ forEachStatefulTask(tasks.standbyTasks(), subtopologies, task -> {
+ if (warmupPlan.borrowedMigrations().contains(task)
+ && !holdsCopyOf(currentAssignment, memberId, task)) {
+ borrowPaidBy.merge(task, memberId, (left, right) ->
left.compareTo(right) <= 0 ? left : right);
Review Comment:
This checks only the destination member, but task uniqueness and standby
handoff are process-scoped. If a target standby moves from `memberC1` to
sibling `memberC2` while another process borrows the task, `memberC2` is
incorrectly selected to pay for the borrow even though process C already has
that copy and the reconciler will serialize the sibling move. Since `assemble`
also drops `memberC1` according to the target assignment, process C loses its
existing standby until the active migration completes. Resolve the destination
member's process and treat a copy on any sibling of that process as already
held (consistent with `CurrentAssignmentBuilder.isUnreleasedStandbyTask`).
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerTest.java:
##########
@@ -1348,6 +1352,494 @@ public void
shouldPlanNothingWhenTheWarmupBudgetIsZero() {
assertEquals(Set.of(), plan.parkedMigrations());
}
+ //
---------------------------------------------------------------------------------------------------------------
+ // filterStandbys
+ //
---------------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void shouldWithholdAStandbyOnAProcessThatStillRunsTheTaskAsActive()
{
+ // The swap shape: the assignor moves the active to memberB and leaves
a standby behind on memberA. While the
+ // migration is staged the task keeps running on memberA, so the
standby cannot be placed there as well.
+ final Map<String, StreamsGroupMember> members = Map.of(
+ "memberA", member("memberA", "processA",
mkTasksTuple(TaskRole.ACTIVE, mkTasks(STATEFUL, 0))),
+ "memberB", member("memberB", "processB", TasksTuple.EMPTY)
+ );
+ final Map<String, TasksTuple> targetAssignment = Map.of(
+ "memberA", mkTasksTuple(TaskRole.STANDBY, mkTasks(STATEFUL, 0)),
+ "memberB", mkTasksTuple(TaskRole.ACTIVE, mkTasks(STATEFUL, 0))
+ );
+
+ assertEquals(Map.of("memberA", Set.of(STATEFUL_0)), filter(members,
targetAssignment, Map.of(), 1));
+ }
+
+ @Test
+ public void
shouldEmitTheStandbyOnTheMemberGrantingTheActiveAwayInTheSameStep() {
+ // The efficient half of the swap: memberA hands the active over and
keeps a standby in its place, which the
+ // client does by relabelling the task it already has. Nothing is
staged, so no rule holds the placement
+ // back, and the relabel happens now rather than a step later when
that state is already gone.
+ final Map<String, StreamsGroupMember> members = Map.of(
+ "memberA", member("memberA", "processA",
mkTasksTuple(TaskRole.ACTIVE, mkTasks(STATEFUL, 0))),
+ "memberB", member("memberB", "processB",
mkTasksTuple(TaskRole.STANDBY, mkTasks(STATEFUL, 0)))
+ );
+ final Map<String, TasksTuple> targetAssignment = Map.of(
+ "memberA", mkTasksTuple(TaskRole.STANDBY, mkTasks(STATEFUL, 0)),
+ "memberB", mkTasksTuple(TaskRole.ACTIVE, mkTasks(STATEFUL, 0))
+ );
+ // memberB's standby is caught up, so the migration is granted rather
than staged.
+ final Map<String, MemberTaskOffsets> taskOffsets = Map.of("memberB",
offsets(100, 100));
+
+ assertEquals(Map.of(), filter(members, targetAssignment, taskOffsets,
1));
+ }
+
+ @Test
+ public void shouldEmitAStandbyOnASiblingOfTheMemberGrantingTheActiveAway()
{
+ // memberA2 would be a second copy on processA until memberA1's
hand-over finishes, and it does have to wait
+ // for it -- but in the reconciler, which holds the placement back
while the process still runs the task.
+ // Withholding it here as well would only add an epoch.
+ final Map<String, StreamsGroupMember> members = Map.of(
+ "memberA1", member("memberA1", "processA",
mkTasksTuple(TaskRole.ACTIVE, mkTasks(STATEFUL, 0))),
+ "memberA2", member("memberA2", "processA", TasksTuple.EMPTY),
+ "memberB", member("memberB", "processB",
mkTasksTuple(TaskRole.STANDBY, mkTasks(STATEFUL, 0)))
+ );
+ final Map<String, TasksTuple> targetAssignment = Map.of(
+ "memberA1", TasksTuple.EMPTY,
+ "memberA2", mkTasksTuple(TaskRole.STANDBY, mkTasks(STATEFUL, 0)),
+ "memberB", mkTasksTuple(TaskRole.ACTIVE, mkTasks(STATEFUL, 0))
+ );
+ final Map<String, MemberTaskOffsets> taskOffsets = Map.of("memberB",
offsets(100, 100));
+
+ assertEquals(Map.of(), filter(members, targetAssignment, taskOffsets,
1));
+ }
+
+ @Test
+ public void shouldEmitAStandbyBlockedOnlyByAPendingRevocation() {
+ // The filter reads the tasks members have been granted, never the
ones they were told to give up: indexing
+ // revocations for this rule alone would duplicate what the reconciler
already enforces, which refuses to
+ // grant a role for a task the process still physically holds. So this
is emitted and the hand-over
+ // serializes itself, at the cost of an extra heartbeat or two before
the group settles.
+ final Map<String, StreamsGroupMember> members = Map.of(
+ "memberA", member(
+ "memberA",
+ "processA",
+ TasksTuple.EMPTY,
+ mkTasksTuple(TaskRole.ACTIVE, mkTasks(STATEFUL, 0))
+ )
+ );
+ final Map<String, TasksTuple> targetAssignment = Map.of(
+ "memberA", mkTasksTuple(TaskRole.STANDBY, mkTasks(STATEFUL, 0))
+ );
+
+ assertEquals(Map.of(), filter(members, targetAssignment, Map.of(), 1));
+ }
+
+ @Test
+ public void shouldWithholdAStandbyOnTheProcessAMigrationIsStagedOn() {
+ // The placement rule 1 protects is the one the staged migration
makes: the task runs on memberB for this
+ // step, so the target assignment's standby of it cannot land on
memberB's process as well. Nothing holds it
Review Comment:
The sentence reads “Nothing holds it task here”; change the final “it” to
“the.”
--
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]
