lucasbru commented on code in PR #22658:
URL: https://github.com/apache/kafka/pull/22658#discussion_r3491581421
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4380,6 +4380,79 @@ 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,
+ final Map<String, MemberTaskOffsets> taskOffsets,
+ final int numWarmupReplicas,
+ final long acceptableRecoveryLag
+ ) {
+ return targetAssignment.getOrDefault(member.memberId(),
TasksTuple.EMPTY);
+ }
+
+ /**
+ * Returns the group's persisted target assignment, with the relabelling
of a replacing static
+ * member applied for the refiner.
+ * <p>
+ * When a static member rejoins with a new member id, the record that
relabels its target
+ * assignment from the old to the new member id is queued during the
heartbeat but not replayed
+ * into the in-memory assignment until afterwards. The persisted map
therefore still keys the
+ * assignment under the old member id at this point. We mirror the pending
relabelling in the
+ * view the refiner consumes — move the assignment to the new member id
and drop the stale old
+ * entry — using the same relabelling-aware lookup as {@link
StreamsGroup#targetAssignment(String, Optional)}.
+ */
+ private static Map<String, TasksTuple>
targetAssignmentWithStaticRelabelling(
+ StreamsGroup group,
+ Optional<StreamsGroupMember> updatedMember
+ ) {
+ Map<String, TasksTuple> targetAssignment = group.targetAssignment();
+ if (updatedMember.isEmpty() ||
updatedMember.get().instanceId().isEmpty()) {
Review Comment:
what does it mean for updatedMember to be empty here?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2258,7 +2253,18 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
currentAssignmentConfigs
);
- // 5. Reconcile the member's assignment with the target assignment if
the member is not
+ // 4b. Refine the target assignment into the intermediate assignment
(with warm-up tasks) the member should be
+ // reconciled toward. Runs on every heartbeat, before reconciliation.
No-op for now.
+ TasksTuple refinedTarget = refine(
+ updatedMember,
+ updateTargetAssignmentResult.targetAssignment,
+ group.targetAssignment(),
+ group.taskOffsets(),
+ config.streamsGroupNumWarmupReplicas(),
Review Comment:
This passes the broker default, ignoring group overrides.
--
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]
