lucasbru commented on code in PR #23794:
URL: https://github.com/apache/kafka/pull/23794#discussion_r4413955920
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -344,6 +335,40 @@ static TaskDecisions analyzeTasks(
return new TaskDecisions(List.copyOf(stagedMigrations),
List.copyOf(grantedTasks));
}
+ /**
+ * Which member the task keeps running on until the target owner is warm.
'Empty' hands the task to the target
+ * owner in this step instead, cold unless its own process still has the
state on disk.
+ */
+ private static Optional<String> keepRunningOn(
+ final CurrentAssignmentIndex currentAssignment,
+ final TaskId task,
+ final ActiveHolder holder,
+ final String targetProcessId,
+ final Map<TaskId, Set<String>> targetStandbyHolders,
+ final Map<String, Integer> heartbeatVersions,
+ final Map<String, ProcessLoad> processLoad
+ ) {
+ if (holder != null && holder.hot()) {
+ return Optional.of(holder.memberId());
+ }
+ if (holder == null && onDisk(currentAssignment, targetProcessId,
task)) {
+ // The target owner's process still has this task's state on disk
and reopens it, so the task goes
+ // there rather than onto a copy holder: the usual way into this
branch is a member that restarted
+ // inside the session timeout and got its own tasks back, where
that state is exact. A stale disk
+ // state is not told apart from a fresh one -- a process reports
an end offset only for a task it
+ // is restoring -- and the copy-holder detour would cost a second
restore, a slot and a hand-over.
+ return Optional.empty();
+ }
+ final Optional<String> copy = bestCopyToPromote(currentAssignment,
task, targetStandbyHolders, processLoad);
+ // Only for an orphan whose target owner's process holds no copy of
the task. A copy there is warmer than one
+ // of unknown lag elsewhere, and is what a hand-over that is already
under way completes with.
+ if (copy.isEmpty() && holder == null &&
findCopyOnProcess(currentAssignment, task, targetProcessId).isEmpty()) {
+ return unobservableStandbyToPromote(
+ currentAssignment, task, targetProcessId,
targetStandbyHolders, heartbeatVersions, processLoad);
+ }
+ return copy;
Review Comment:
Went with the variant from your other comment in 2e445a04b5: keepRunningOn
now ends with `bestCopyToPromote(...).or(() ->
unobservableStandbyToPromote(...))`, and the checks moved into the method.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -371,6 +396,42 @@ private static Optional<String> bestCopyToPromote(
.map(TaskCopy::memberId);
}
+ /**
+ * The standby task of an orphan that is promoted to run the task although
the coordinator cannot tell how far
+ * behind it is, because its holder does not report its restore progress:
a member that is not
+ * {@link WarmupSupport#SUPPORTED}. Only a standby task that the target
assignment also gives to that member, and
+ * that is not on the target owner's process, is promoted. The target
owner's process is {@code null} if the
+ * target owner has left the group.
+ *
+ * <p>This is the one place where a copy of unknown lag is acted on.
Elsewhere, the alternative to promoting a
+ * lagging copy is keeping a running owner, here it is a restore from
scratch on the target owner next to a copy
+ * that would have to restore anyway: a stale copy is restored as an
active task by the temporary owner before it
+ * processes anything, which is the restore the target assignment demands
of it as a standby holder, and a copy
+ * that is close resumes processing at once. When the target owner's
warm-up task has caught up, the hand-over
+ * demotes the temporary owner to its standby task in place.
Review Comment:
Rewrote it in 2e445a04b5. It now says what the method picks, why that beats
restoring from scratch on the target owner, and the conditions under which it
applies. Let me know if it still reads too dense.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -236,7 +236,9 @@ static Map<String, ProcessLoad> indexProcessLoad(
* process: (1) either the old owner of the task keeps running it, or (2)
caught-up copy which we temporarily
* promote as active is running it.
*
- * <p>Only an active-running or caught-up copy is ever promoted.
+ * <p>Only an active-running or caught-up copy is ever promoted, with one
exception: the standby task of a task
+ * that no member runs as an active task, on a member that does not report
its restore progress, see
Review Comment:
Switched to "holds" in 5aa5543c70, together with the line above.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -371,6 +392,45 @@ private static Optional<String> bestCopyToPromote(
.map(TaskCopy::memberId);
}
+ /**
+ * Picks a standby task to run a task that no member runs as an active
task, while the target owner warms it up,
+ * even though the standby's holder does not report how far it has
restored: the holder heartbeats with version 0,
+ * or has not heartbeated since the coordinator loaded the group.
+ *
+ * <p>Without this, the target owner would restore the task from scratch
while nobody runs it. On the standby's
+ * holder, the task runs sooner: a standby that is behind catches up
before it processes, which it would have to
+ * do as a standby anyway, and one that is close processes right away.
Once the target owner's warm-up task has
+ * caught up, the task moves there and the holder keeps it as a standby
task.
+ *
+ * <p>It applies only if no member runs the task as an active task, the
target owner's process holds no copy of
+ * the task, which the target owner would otherwise continue from, and the
target assignment keeps the standby
+ * task on its holder. If several standby tasks qualify, the one on the
least loaded process is picked.
+ *
+ * @param targetProcessId The process of the target owner, or {@code null}
if the target owner has left the group.
+ */
Review Comment:
Took your version in 5aa5543c70.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -371,6 +396,42 @@ private static Optional<String> bestCopyToPromote(
.map(TaskCopy::memberId);
}
+ /**
+ * The standby task of an orphan that is promoted to run the task although
the coordinator cannot tell how far
+ * behind it is, because its holder does not report its restore progress:
a member that is not
+ * {@link WarmupSupport#SUPPORTED}. Only a standby task that the target
assignment also gives to that member, and
+ * that is not on the target owner's process, is promoted. The target
owner's process is {@code null} if the
+ * target owner has left the group.
+ *
+ * <p>This is the one place where a copy of unknown lag is acted on.
Elsewhere, the alternative to promoting a
+ * lagging copy is keeping a running owner, here it is a restore from
scratch on the target owner next to a copy
+ * that would have to restore anyway: a stale copy is restored as an
active task by the temporary owner before it
+ * processes anything, which is the restore the target assignment demands
of it as a standby holder, and a copy
+ * that is close resumes processing at once. When the target owner's
warm-up task has caught up, the hand-over
+ * demotes the temporary owner to its standby task in place.
+ *
+ * <p>Among several candidates, the least loaded process wins, without
comparing lags.
+ */
+ private static Optional<String> unobservableStandbyToPromote(
+ final CurrentAssignmentIndex currentAssignment,
+ final TaskId task,
+ final String targetProcessId,
+ final Map<TaskId, Set<String>> targetStandbyHolders,
+ final Map<String, Integer> heartbeatVersions,
+ final Map<String, ProcessLoad> processLoad
+ ) {
+ final Set<String> designatedStandbyHolders =
targetStandbyHolders.getOrDefault(task, Set.of());
+ return currentAssignment.taskCopies().getOrDefault(task,
List.of()).stream()
+ .filter(copy -> copy.role() == TaskRole.STANDBY
+ && !copy.processId().equals(targetProcessId)
Review Comment:
Agreed, with the target-process check inside the method the filter was
redundant. Removed in 2e445a04b5.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -371,6 +396,42 @@ private static Optional<String> bestCopyToPromote(
.map(TaskCopy::memberId);
}
+ /**
+ * The standby task of an orphan that is promoted to run the task although
the coordinator cannot tell how far
+ * behind it is, because its holder does not report its restore progress:
a member that is not
+ * {@link WarmupSupport#SUPPORTED}. Only a standby task that the target
assignment also gives to that member, and
+ * that is not on the target owner's process, is promoted. The target
owner's process is {@code null} if the
+ * target owner has left the group.
+ *
+ * <p>This is the one place where a copy of unknown lag is acted on.
Elsewhere, the alternative to promoting a
+ * lagging copy is keeping a running owner, here it is a restore from
scratch on the target owner next to a copy
+ * that would have to restore anyway: a stale copy is restored as an
active task by the temporary owner before it
+ * processes anything, which is the restore the target assignment demands
of it as a standby holder, and a copy
+ * that is close resumes processing at once. When the target owner's
warm-up task has caught up, the hand-over
+ * demotes the temporary owner to its standby task in place.
+ *
+ * <p>Among several candidates, the least loaded process wins, without
comparing lags.
+ */
+ private static Optional<String> unobservableStandbyToPromote(
Review Comment:
Done in 2e445a04b5, pretty much as you suggested. Both callers now go
through the same `.or(...)`. I also added the three tests from your review
comment, and checked that each fails under its mutation.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -1250,9 +1310,11 @@ record TaskCopy(String memberId, String processId,
TaskRole role, boolean caught
* The task being migrated.
* @param currentOwner
* The member the task runs on for now: the member holding it as an
active task, or, when that member is
- * too far behind to run it or there is none, the caught-up copy
holder promoted to run it. When the target
- * assignment names a member the group no longer has, it can also
be a holder that is still restoring the
- * task, which is better than taking the task away from a restore
that is part-way through.
+ * too far behind to run it or there is none, the caught-up copy
holder promoted to run it. For a task that
+ * no member runs as an active task, it can also be a standby
holder that does not report its restore
Review Comment:
Done in 5aa5543c70.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -236,7 +236,9 @@ static Map<String, ProcessLoad> indexProcessLoad(
* process: (1) either the old owner of the task keeps running it, or (2)
caught-up copy which we temporarily
* promote as active is running it.
*
- * <p>Only an active-running or caught-up copy is ever promoted.
+ * <p>Only an active-running or caught-up copy is ever promoted, with one
exception: the standby task of a task
Review Comment:
Done in 5aa5543c70. I kept "of a task that" on the next line so the sentence
still reads with your change below.
--
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]
