KAFKA-20665: Bump group epoch for warm-up refinement steps#22748
Conversation
Replace the streams-heartbeat bumpGroupEpoch boolean with an AssignmentUpdate enum (NONE/RECOMPUTE/REFINE). RECOMPUTE keeps today's behavior (re-run the assignor for a new final target). REFINE only advances the assignment epoch of the unchanged final target via the target-assignment metadata record, without re-running the assignor, so members re-reconcile toward the next intermediate (warm-up) step. A REFINE bump is triggered by hasHotWarmupTask(): a caught-up warm-up (lag within acceptable.recovery.lag). Each refinement step needs its own assignment epoch because members only pick up a changed assignment when the epoch advances, and a new assignment epoch requires a group-epoch bump.
aliehsaeedii
left a comment
aliehsaeedii
left a comment
There was a problem hiding this comment.
Thanks @mjsax. I tried to undertand the code and left some comments/questions.
| * @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) { |
There was a problem hiding this comment.
What about test coverage? Maybe we can add GroupMetadataManagerTest cases for at least:
- a
STABLEgroup with a caught-up warm-up task bumps the group epoch and writes only theStreamsGroupTargetAssignmentMetadatarecord (assignor not invoked, per-member target records untouched); - no bump when the group isn't
STABLE, when the lag exceedsacceptable.recovery.lag, or when offsets/end-offsets haven't been reported for the task; - the boundary case
endOffset - offset == acceptableRecoveryLag.
| Optional<List<Status>> returnedStatus, | ||
| Map<String, String> assignmentConfigs | ||
| Map<String, String> assignmentConfigs, | ||
| boolean refineOnly |
There was a problem hiding this 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))
| // Actually bump the group epoch | ||
| int groupEpoch = group.groupEpoch(); | ||
| if (bumpGroupEpoch) { | ||
| if (assignmentUpdate != AssignmentUpdate.NONE) { |
There was a problem hiding this 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!
| assignmentUpdate = AssignmentUpdate.RECOMPUTE; | ||
| } | ||
|
|
||
| // Maybe check if warmup task are hot, and the assignment can be refined furtther |
There was a problem hiding this comment.
| // Maybe check if warmup task are hot, and the assignment can be refined furtther | |
| // Maybe check if warmup tasks are hot, and the assignment can be refined further |
| // 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())); |
There was a problem hiding this 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?!
| return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(), updatedMembersAndTargetAssignment.targetAssignment()); | ||
| } | ||
|
|
||
| if (refineOnly) { |
There was a problem hiding this comment.
Could we add && group.groupEpoch() >= group.assignmentEpoch() to the condition to rule out target assignment epoch updates when there is a pending batched or offloaded assignment? The caller guarantees this condition right now but it looks easy to break and introduce a bug.
3 participants
Replace the streams-heartbeat bumpGroupEpoch boolean with an
AssignmentUpdate enum (NONE/RECOMPUTE/REFINE). RECOMPUTE keeps today's
behavior (re-run the assignor for a new final target). REFINE only
advances the assignment epoch of the unchanged final target via the
target-assignment metadata record, without re-running the assignor, so
members re-reconcile toward the next intermediate (warm-up) step.
A REFINE bump is triggered by hasHotWarmupTask(): a caught-up warm-up
(lag within acceptable.recovery.lag). Each refinement step needs its own
assignment epoch because members only pick up a changed assignment when
the epoch advances, and a new assignment epoch requires a group-epoch
bump.
Reviewers: Alieh Saeedi asaeedi@confluent.io, Sean Quah
squah@confluent.io