lucasbru commented on code in PR #23488:
URL: https://github.com/apache/kafka/pull/23488#discussion_r4044860908
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -421,36 +969,83 @@ private static boolean isStateful(
* placement to preserve.
* @param taskCopies
* The standby and warm-up holders of each task.
+ * @param onDiskByProcess
+ * Per process, the tasks it reports state for while holding no
copy of them: state in its state directory
+ * that no role it has been granted accounts for. How far behind
that state is cannot be measured, because
+ * a member reports an end offset only for a task it is restoring.
A process with no such task has no
+ * entry.
*/
record CurrentAssignmentIndex(
Map<TaskId, ActiveHolder> activeHolder,
- Map<TaskId, List<TaskCopy>> taskCopies
+ Map<TaskId, List<TaskCopy>> taskCopies,
+ Map<String, Set<TaskId>> onDiskByProcess
) {
}
/**
- * The member holding a task as an active task, and whether it is
processing the task or still restoring it.
+ * How much stateful work a process is carrying, for the warm-up funding
order.
+ *
+ * <p>The count and the divisor are kept separately because the funding
order needs the quotient while the
+ * accounting needs the count: each warm-up task funded within one pass
raises its target process's load before
+ * the next pick is made, which {@link #loadWith(int)} does without
disturbing the index itself.
*
- * <p>Only a member that is <em>processing</em> a task has something a
staged migration could protect, so the two
- * are kept apart rather than collapsed into a plain member ID. The
identity is still needed for a member that is
- * only restoring, though: it is what tells the case analysis the task is
already in the right place, and what lets
- * a task be kept where it is when the target assignment names a member
the group no longer has.
+ * @param statefulTaskCount
+ * How many stateful tasks the process has been granted, counting
every role. See
+ * {@link #indexProcessLoad} for why stateless tasks are left out.
+ * @param memberCount
+ * How many members the process runs. Each member is one stream
thread, so this is the process's
+ * capacity for running tasks.
+ */
+ record ProcessLoad(int statefulTaskCount, int memberCount) {
+
+ double load() {
+ return loadWith(0);
+ }
+
+ /**
+ * The load this process would carry with the given number of warm-up
tasks added to it.
+ */
+ double loadWith(final int newWarmupTasks) {
+ return (double) (statefulTaskCount + newWarmupTasks) / memberCount;
+ }
+ }
+
+ /**
+ * The member holding a task as an active task, and how usable the state
it holds is.
+ *
+ * <p>Only a {@link #hot()} holder has state a staged migration can use,
so the two flags it is derived
+ * from are recorded alongside the member ID.
+ *
+ * <p>The identity matters even for a holder that is not hot: it is what
tells the case analysis the task is
+ * already in the right place, and what lets a task be kept where it is
when the target assignment names a member
+ * the group no longer has.
*
* @param memberId
* The member holding the task.
- * @param processing
- * Whether the member is processing the task, as opposed to still
restoring it. See {@link #isRestoring}
+ * @param restoring
+ * Whether the member is still restoring the task, as opposed to
processing it. See {@link #isRestoring}
* for how this is determined, and for why a member the coordinator
has not heard from reads as processing.
+ * @param caughtUp
+ * Whether the member has restored the task to within {@code
acceptable.recovery.lag}.
+ * Should only be checked if the task is not {@code restoring}.
Review Comment:
I think this doc has it backwards - caughtUp is only meaningful while
restoring is true (once a member stops restoring it stops reporting offsets for
the task, so caughtUp would read as false). hot() relies on the short-circuit
`!restoring || caughtUp` to be correct today, but as written this would mislead
anyone who reads caughtUp on its own.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -221,46 +262,112 @@ static TaskDecisions analyzeTasks(
// was removed from the group in the meantime. For this case, all
previously owned tasks of this member
// (which did not get move to a new owner) will be "dandling"
which will be fixed by the next assignor run.
// Furthermore, we stage all tasks the assignor moves to this
member on their old owners to keep them
- // "online".
+ // "online". A task nobody holds as an active task is put on a
caught-up copy holder instead, which is the
+ // only way it runs at all before the next assignor run.
final StreamsGroupMember targetMember = members.get(targetOwner);
if (targetMember == null) {
- if (holder != null) {
- stagedMigrations.add(new StagedMigration(
- task,
- holder.memberId(),
- targetOwner,
- Optional.empty(),
- Optional.empty()
- ));
- }
+ final Optional<String> currentOwner = holder != null
+ ? Optional.of(holder.memberId())
+ : bestCopyToPromote(currentAssignment, task,
targetStandbyHolders, processLoad);
+ currentOwner.ifPresent(owner -> stagedMigrations.add(
+ stagedMigration(currentAssignment, task, owner,
targetOwner, Optional.empty())));
continue;
}
final String targetProcessId = targetMember.processId();
+ final Optional<String> currentProcessId =
Optional.ofNullable(holder)
+ .map(activeHolder ->
members.get(activeHolder.memberId()).processId());
- // The task moves now, for any of three reasons.
- // 1. Nobody holds it.
- // 2. Somebody holds it but is still restoring it.
- // 3. Somebody is processing it and the target owner is caught up
- if (holder == null
- || !holder.processing()
- || isReady(currentAssignment, task,
members.get(holder.memberId()).processId(), targetProcessId)) {
+ if (isReady(currentAssignment, task, currentProcessId,
targetProcessId)) {
grantedTasks.add(new TaskGrant(task, targetOwner));
continue;
}
- stagedMigrations.add(new StagedMigration(
- task,
- holder.memberId(),
- targetOwner,
- Optional.of(targetProcessId),
- findCopyOnProcess(currentAssignment, task, targetProcessId)
- ));
+ // Which member the task keeps running on until the target owner
is warm. Empty means nobody has state
+ // worth keeping the task on, so the target owner takes it over
cold.
+ final Optional<String> currentOwner;
+ if (holder != null && holder.hot()) {
+ currentOwner = Optional.of(holder.memberId());
+ } else if (holder == null && onDisk(currentAssignment,
targetProcessId, task)) {
+ // The target owner's process left this task's state on disk
and can reopen it. How far behind that
+ // state is cannot be measured -- a member reports an end
offset only for a task it is restoring --
Review Comment:
Another member could report the end offset (because it's actively
restoring). This is actually likely to be the case. Could we measure the lag
that way?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -221,46 +262,112 @@ static TaskDecisions analyzeTasks(
// was removed from the group in the meantime. For this case, all
previously owned tasks of this member
// (which did not get move to a new owner) will be "dandling"
which will be fixed by the next assignor run.
// Furthermore, we stage all tasks the assignor moves to this
member on their old owners to keep them
- // "online".
+ // "online". A task nobody holds as an active task is put on a
caught-up copy holder instead, which is the
+ // only way it runs at all before the next assignor run.
final StreamsGroupMember targetMember = members.get(targetOwner);
if (targetMember == null) {
- if (holder != null) {
- stagedMigrations.add(new StagedMigration(
- task,
- holder.memberId(),
- targetOwner,
- Optional.empty(),
- Optional.empty()
- ));
- }
+ final Optional<String> currentOwner = holder != null
+ ? Optional.of(holder.memberId())
+ : bestCopyToPromote(currentAssignment, task,
targetStandbyHolders, processLoad);
+ currentOwner.ifPresent(owner -> stagedMigrations.add(
+ stagedMigration(currentAssignment, task, owner,
targetOwner, Optional.empty())));
continue;
}
final String targetProcessId = targetMember.processId();
+ final Optional<String> currentProcessId =
Optional.ofNullable(holder)
+ .map(activeHolder ->
members.get(activeHolder.memberId()).processId());
- // The task moves now, for any of three reasons.
- // 1. Nobody holds it.
- // 2. Somebody holds it but is still restoring it.
- // 3. Somebody is processing it and the target owner is caught up
- if (holder == null
- || !holder.processing()
- || isReady(currentAssignment, task,
members.get(holder.memberId()).processId(), targetProcessId)) {
+ if (isReady(currentAssignment, task, currentProcessId,
targetProcessId)) {
grantedTasks.add(new TaskGrant(task, targetOwner));
continue;
}
- stagedMigrations.add(new StagedMigration(
- task,
- holder.memberId(),
- targetOwner,
- Optional.of(targetProcessId),
- findCopyOnProcess(currentAssignment, task, targetProcessId)
- ));
+ // Which member the task keeps running on until the target owner
is warm. Empty means nobody has state
+ // worth keeping the task on, so the target owner takes it over
cold.
+ final Optional<String> currentOwner;
+ if (holder != null && holder.hot()) {
+ currentOwner = Optional.of(holder.memberId());
+ } else if (holder == null && onDisk(currentAssignment,
targetProcessId, task)) {
+ // The target owner's process left this task's state on disk
and can reopen it. How far behind that
+ // state is cannot be measured -- a member reports an end
offset only for a task it is restoring --
+ // so this takes precedence over promoting a caught-up copy,
because the common way a task ends up
Review Comment:
Is that right? For a caught up copy, we know that we are not behind, for the
on disk task, we don't.
--
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]
