squah-confluent commented on code in PR #22658:
URL: https://github.com/apache/kafka/pull/22658#discussion_r3481103306
##########
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:
I'm curious about why we would want the target assignment both before and
after the current heartbeat's assignment run. Naively I would expect the latest
target assignment only.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4309,15 +4304,12 @@ private UpdateTargetAssignmentResult<TasksTuple>
maybeUpdateStreamsTargetAssignm
.setStatusDetail("Assignment delayed due to the configured
initial rebalance delay.")
));
- return new UpdateTargetAssignmentResult<>(
- group.assignmentEpoch(),
- TasksTuple.EMPTY
- );
+ return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(),
Map.of());
}
if (group.assignmentEpoch() >= groupEpoch) {
// The assignment is up to date.
- return
UpdateTargetAssignmentResult.fromLastTargetAssignment(group, updatedMember);
+ return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(),
group.targetAssignment());
Review Comment:
There's a huge footgun here to watch out for. When a static member rejoins,
we relabel the assignment entry in the target assignment with the new member
id. `group.targetAssignment()` does not include the relabelling until after the
heartbeat.
I ran into this problem in a different way while working on assignment
offload. The solution I ended up with is in
https://github.com/apache/kafka/pull/22510/changes#diff-00f0f81cf13e66781777d94f7d2e68a581663385c37e98792507f2294c91bb09.
--
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]
