aliehsaeedii commented on code in PR #22748:
URL: https://github.com/apache/kafka/pull/22748#discussion_r3518709403
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4326,7 +4388,8 @@ private UpdateTargetAssignmentResult<Map<String,
TasksTuple>> maybeUpdateStreams
CoordinatorMetadataImage metadataImage,
List<CoordinatorRecord> records,
Optional<List<Status>> returnedStatus,
- Map<String, String> assignmentConfigs
+ Map<String, String> assignmentConfigs,
+ boolean refineOnly
Review Comment:
nit: the other call site (`computeDelayedTargetAssignment`) now passes a
bare `false`. Passing the `AssignmentUpdate` enum instead of a boolean would
read better at both call sites (`..., AssignmentUpdate.RECOMPUTE)` vs `...,
false)`)
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4307,6 +4321,52 @@ private UpdateTargetAssignmentResult<Assignment>
maybeUpdateTargetAssignment(
}
}
+ /**
+ * How a streams-group heartbeat updates the group epoch and target
assignment.
+ * <ul>
+ * <li>{@code NONE} - nothing changed; the group epoch is not
bumped.</li>
+ * <li>{@code RECOMPUTE} - the group changed (membership, topology or
configuration); bump the group epoch and re-run the
+ * assignor to produce a new final target assignment.</li>
+ * <li>{@code REFINE} - a warm-up task has caught up; bump the group
epoch to advance the in-memory refinement one step
+ * WITHOUT re-running the assignor (the final target assignment is
unchanged).</li>
+ * </ul>
+ */
+ private enum AssignmentUpdate {
+ NONE,
+ RECOMPUTE,
+ REFINE
+ }
+
+ /**
+ * Returns whether at least one currently-assigned warm-up task has caught
up, i.e. its lag (reported end-offset minus reported
+ * offset) is within the acceptable recovery lag and it is therefore ready
to be promoted to active.
+ *
+ * @param group The streams group.
+ * @param acceptableRecoveryLag The lag threshold below which a warm-up is
considered caught up.
+ * @return true if at least one assigned warm-up task is caught up.
+ */
+ private boolean hasHotWarmupTask(StreamsGroup group, long
acceptableRecoveryLag) {
Review Comment:
What about test coverage? Maybe we can add `GroupMetadataManagerTest` cases
for at least:
- a `STABLE` group with a caught-up warm-up task bumps the group epoch and
writes only the `StreamsGroupTargetAssignmentMetadata` record (assignor not
invoked, per-member target records untouched);
- no bump when the group isn't `STABLE`, when the lag exceeds
`acceptable.recovery.lag`, or when offsets/end-offsets haven't been reported
for the task;
- the boundary case `endOffset - offset == acceptableRecoveryLag`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2196,21 +2201,29 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
}
// We validated a topology that was not validated before, so bump the
group epoch as we may have to reassign tasks.
if (validatedTopologyEpoch != group.validatedTopologyEpoch()) {
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
}
// Check if assignment configurations have changed
Map<String, String> currentAssignmentConfigs =
streamsGroupAssignmentConfigs(groupId);
Map<String, String> storedAssignmentConfigs =
group.lastAssignmentConfigs();
- if (!bumpGroupEpoch &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
+ if (assignmentUpdate == AssignmentUpdate.NONE &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
log.info("[GroupId {}][MemberId {}] Assignment configurations
changed to {}. Triggering rebalance.",
groupId, memberId, currentAssignmentConfigs);
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
+ }
+
+ // Maybe check if warmup task are hot, and the assignment can be
refined furtther
Review Comment:
```suggestion
// Maybe check if warmup tasks are hot, and the assignment can be
refined further
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2196,21 +2201,29 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
}
// We validated a topology that was not validated before, so bump the
group epoch as we may have to reassign tasks.
if (validatedTopologyEpoch != group.validatedTopologyEpoch()) {
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
}
// Check if assignment configurations have changed
Map<String, String> currentAssignmentConfigs =
streamsGroupAssignmentConfigs(groupId);
Map<String, String> storedAssignmentConfigs =
group.lastAssignmentConfigs();
- if (!bumpGroupEpoch &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
+ if (assignmentUpdate == AssignmentUpdate.NONE &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
log.info("[GroupId {}][MemberId {}] Assignment configurations
changed to {}. Triggering rebalance.",
groupId, memberId, currentAssignmentConfigs);
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
+ }
+
+ // Maybe check if warmup task are hot, and the assignment can be
refined furtther
+ if (assignmentUpdate == AssignmentUpdate.NONE
+ && group.state() == StreamsGroup.StreamsGroupState.STABLE
+ && hasHotWarmupTask(group,
streamsGroupAcceptableRecoveryLag(groupId))
+ ) {
+ assignmentUpdate = AssignmentUpdate.REFINE;
}
// Actually bump the group epoch
int groupEpoch = group.groupEpoch();
- if (bumpGroupEpoch) {
+ if (assignmentUpdate != AssignmentUpdate.NONE) {
Review Comment:
I don't understand this. I assumed bumping must happen when the refiner's
output for some member differs from that member's current assignment?!
You change `assignmentUpdate = AssignmentUpdate.REFINE;` above under a
condition that it means "some warm-up is caught up" and NOT "the refiner would
actually change the intermediate assignment"! I don't get how these two are
equal and if they really are (under any circumstance and considering all edge
cases)
BTW, `refine()` currently returns the unchanged target, so nothing ever
promotes the warm-up. If this PR merges standalone and anything puts a warm-up
into an assignment, the sequence becomes: hot → bump → nothing changes → STABLE
again → hot → bump → … forever. I know, today that's unreachable (no built-in
assignor emits warm-ups), that's why you considered it as safe to merge for now!
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4358,6 +4421,14 @@ private UpdateTargetAssignmentResult<Map<String,
TasksTuple>> maybeUpdateStreams
return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(),
updatedMembersAndTargetAssignment.targetAssignment());
}
+ if (refineOnly) {
+ // A refinement step (warm-up promotion) only advances the
assignment epoch of the unchanged final target so that members
+ // re-reconcile toward the next in-memory intermediate assignment.
The assignor is not run and the per-member target
+ // assignment records are left untouched; only the (small)
target-assignment metadata record carrying the epoch is written.
+
records.add(newStreamsGroupTargetAssignmentMetadataRecord(group.groupId(),
groupEpoch, time.milliseconds()));
Review Comment:
Why `time.milliseconds()` here and not `group.assignmentTimestamp()`? The
assignor didn't run in a refinement step, but writing a fresh timestamp
restarts the assignment-interval?!
--
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]
