You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2026/10/13 08:27:24 UTC
[PR] KAFKA-21111 [3/N]: Promote the standby of an orphan on a member that cannot report its progress [kafka]
lucasbru opened a new pull request, #23794:
URL: https://github.com/apache/kafka/pull/23794
This is the third change of the series that makes the warm-up task refiner (KAFKA-20665) compatible with clients that speak version 0 of the streams group heartbeat (see KAFKA-21111), stacked on #23784. The first commit of this branch is that PR; please review only the last commit. I will rebase once #23784 is merged.
A copy of a task on a member that does not report its restore progress can never be observed to be caught up, so it is neither promoted nor protected, like any other copy that is not caught up. There is one exception, for an orphan: a stateful task that nobody holds as an active task, whose target owner can be warmed up but holds no state for it. If the target assignment also names the holder of the unobservable copy as a standby holder of the task, the holder runs the task as temporary owner while the target owner warms up. A holder that heartbeats with version 0, or whose version is not known yet, is such a member.
Here the alternative to promoting a copy of unknown lag is a restore from scratch on the target owner next to a standby task that would have to restore anyway, whereas elsewhere the alternative is keeping a running owner. If the copy turns out to be stale, the temporary owner restores it as an active task before it processes anything, which is the restore the target assignment demands of it as a standby holder anyway. If it is close, the task resumes processing at once. The hand-over then demotes the temporary owner to its standby task in place.
The refiner is not the default yet, so this change is not live. The tests cover the promotion for a holder of version 0 and for one of unknown version, and the cases where it must not apply: a holder that reports being behind, a standby that the target assignment does not keep on its holder, and a target owner that cannot be warmed up.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21111 [3/N]: Promote the standby of an orphan on a member that cannot report its progress [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21111 [3/N]: Promote the standby of an orphan on a member that cannot report its progress [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23794:
URL: https://github.com/apache/kafka/pull/23794#discussion_r4418798883
##########
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:
```suggestion
* no member holds as an active task, on a member that does not report its restore progress, see
```
##########
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:
```suggestion
* <p>Only an active-running or caught-up copy is ever promoted, with one exception: the standby task
```
##########
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:
```suggestion
/**
* The standby holder of the task to promote although its lag is unknown, because the holder is not
* {@link WarmupSupport#SUPPORTED}. Only if no member holds the task as an active task, the target owner's process
* holds no copy of the task, and the target assignment keeps the standby task on its holder; among several, the
* least loaded process wins. A standby task that is behind restores before it processes, which it has to do as a
* standby task anyway.
*
* @param targetProcessId The process of the target owner, or {@code null} if the target owner has left the group.
*/
```
##########
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:
Don't think the current comment is correct.
##########
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:
```suggestion
* no member holds as an active task, it can also be a standby holder that does not report its restore
```
##########
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:
"holds" is more generic, ie, covers both active-running and active-restoring, while "runs" might indicate active-running
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21111 [3/N]: Promote the standby of an orphan on a member that cannot report its progress [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23794:
URL: https://github.com/apache/kafka/pull/23794#discussion_r4256662683
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -356,6 +390,41 @@ 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 is
+ * promoted.
Review Comment:
Updated both in 49747074f2.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21111 [3/N]: Promote the standby of an orphan on a member that cannot report its progress [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #23794:
URL: https://github.com/apache/kafka/pull/23794#discussion_r4256414893
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -356,6 +390,41 @@ 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 is
+ * promoted.
Review Comment:
This new exception makes two existing contracts inaccurate: `analyzeTasks` still says only active-running or caught-up copies are promoted (lines 235–239), and `StagedMigration.currentOwner` still describes promoted copy holders as caught up (lines 1306–1309). Please update both Javadocs to include this unobservable orphan-standby case so callers do not rely on the old invariant.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21111 [3/N]: Promote the standby of an orphan on a member that cannot report its progress [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21111 [3/N]: Promote the standby of an orphan on a member that cannot report its progress [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #23794:
URL: https://github.com/apache/kafka/pull/23794
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org