squah-confluent commented on code in PR #22658:
URL: https://github.com/apache/kafka/pull/22658#discussion_r3491849102
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4392,6 +4380,47 @@ private UpdateTargetAssignmentResult<TasksTuple>
maybeUpdateStreamsTargetAssignm
}
}
+
+ /**
+ * Refines the task assignor's target assignment into the
<em>intermediate</em> assignment that
+ * the reconciler ({@link
org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder}) converges
members toward.
+ * <p>
+ * The intermediate assignment is the current assignment with warm-up
tasks inserted (for later promotion to active,
+ * based on per-member changelog lag), so that a task is moved to a new
owner only once that owner has caught up.
+ * It is held in memory only and is never persisted: the assignor's target
assignment remains the persisted source of
+ * truth (and the intermediate is reconstructed from the persisted
current-assignment records after a coordinator failover).
+ * <p>
+ * The refiner is invoked on every heartbeat, before reconciliation, so it
can react to all the inputs that can change
+ * the intermediate assignment between reassignments — newly reported task
offsets (a warm-up may have become hot),
+ * {@code num.warmup.replicas} / {@code acceptable.recovery.lag} config
changes, and members acknowledging task
+ * revocation/restoration (advancing an in-flight migration).
+ *
+ * @param member
+ * The member to produce the refined (intermediate) assignment for.
+ * @param targetAssignment
+ * All members' target assignments (group context).
+ * @param currentAssignment
+ * The group's current (committed) target assignment, per member
(group context).
+ * @param taskOffsets
+ * The latest per-member changelog offsets/end-offsets reported via
heartbeats (group context).
+ * @param numWarmupReplicas
+ * The configured maximum number of warm-up replicas.
+ * @param acceptableRecoveryLag
+ * The lag at or below which a warm-up is considered caught up and
can be promoted.
+ *
+ * @return The member's intermediate assignment tuple.
+ */
+ private static TasksTuple refine(
+ final StreamsGroupMember member,
+ final Map<String, TasksTuple> targetAssignment,
+ final Map<String, TasksTuple> currentAssignment,
Review Comment:
Thanks for the explanation! I was confused because the current assignment is
stored in `assignedTasks` on each `StreamsGroupMember`. I think we want that
instead?
--
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]
