mjsax commented on code in PR #23794:
URL: https://github.com/apache/kafka/pull/23794#discussion_r4407518108
##########
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:
```suggestion
if (copy.isPresent()) {
return copy;
}
// Only for an orphan whose target owner's process holds no copy of
the task. A copy there is what the
// grant recycles, or what a hand-over already under way completes
with, so the alternative is not a
// restore from scratch
if (holder == null && findCopyOnProcess(currentAssignment, task,
targetProcessId).isEmpty()) {
return unobservableStandbyToPromote(
currentAssignment, task, targetProcessId,
targetStandbyHolders, heartbeatVersions, processLoad);
}
return Optional.empty();
```
##########
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:
Why do we need this filter? If I understand the code correctly, this filter
could only be false, if the assignor puts an active and standby on the same
client, which would be an invalid assignment.
From Claude:
> This filter guards only part of what the exception needs. The rest, an
orphan whose target owner's process holds no copy of any role, is checked by
the caller in keepRunningOn. Could we move both checks into this method as an
early return? Then the method is correct on its own, this filter becomes
unnecessary, and both orphan paths end in the same
bestCopyToPromote(...).or(...).
##########
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:
Reviewing this with the help from Claude reveals that this method does
depend on condition the callee checks first. Might be worth to pull in this
logic into the method to make it self-contained (also related to my comment
below about the filter).
```
private static Optional<String> unobservableStandbyToPromote(...) {
if (currentAssignment.activeHolder().containsKey(task)
|| targetProcessId != null &&
findCopyOnProcess(currentAssignment, task, targetProcessId).isPresent()) {
return Optional.empty();
}
final Set<String> designatedStandbyHolders =
targetStandbyHolders.getOrDefault(task, Set.of());
return currentAssignment.taskCopies().getOrDefault(task,
List.of()).stream()
.filter(copy -> copy.role() == TaskRole.STANDBY
&& designatedStandbyHolders.contains(copy.memberId())
&&
WarmupSupport.of(heartbeatVersions.get(copy.memberId())) !=
WarmupSupport.SUPPORTED)
...
}
```
And
> Then keepRunningOn ends the same way as the target-owner-left branch:
```
return bestCopyToPromote(currentAssignment, task,
targetStandbyHolders, processLoad)
.or(() -> unobservableStandbyToPromote(
currentAssignment, task, targetProcessId,
targetStandbyHolders, heartbeatVersions, processLoad));
```
##########
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:
That's a classic Claude comment. Can you rephrase it to make it easier to
understand. I grok it now, because I work on the subject for months. But if I
cycle back in the future, or somebody less familiar reads this, I am sure it's
hard to impossible to understand.
--
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]
