MINOR: Add runtime support for chained coordinator write operations [WIP] #22739
Open
dajac wants to merge 3 commits into
Open
MINOR: Add runtime support for chained coordinator write operations [WIP] #22739dajac wants to merge 3 commits into
dajac wants to merge 3 commits into
Conversation
…stency check
CurrentAssignmentBuilder currently strips no-longer-subscribed topics from a
member's current assignment based on hasSubscriptionChanged, a boolean diffed
from the member before/after a heartbeat's own mutation. That diff only exists
within the event that computed it and cannot be reconstructed after a replay
boundary, which blocks decomposing the heartbeat into steps that observe
already-replayed state.
Replace it with assignmentIsSubscriptionConsistent(member, resolvedRegularExpressions,
metadataImage): a pure state fact ("is every currently assigned topic still
subscribed to, by name or resolved regex") that can be evaluated at any time,
independent of what changed during the current event. maybeReconcile computes
it once and feeds both its early-return check and the (renamed)
withEnforceSubscriptionConsistency builder input, so behavior is otherwise
unchanged for the paths that already force consistency.
One test's expectations change: when a member's subscription changes but its
current assignment is already empty (nothing to strip), the old diff-based
flag still touched the member solely because updateMemberEpoch() unconditionally
overwrites previousMemberEpoch, producing a redundant record. The new check
correctly treats an empty assignment as trivially consistent, so no record is
emitted. This also avoids incidentally narrowing the previousMemberEpoch retry
tolerance for no reason.
Pre-existing assignment/subscription inconsistencies unrelated to the current
event are now healed as a side effect (previously left stale until something
else touched the member) - the same record content as reconciling that
inconsistency would otherwise produce, just triggered earlier.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
GroupMetadataManager.consumerGroupHeartbeat generates all of a heartbeat's records and its response in a single pass, before any of them are replayed to the state machine. This forces every stage of the handler to thread not-yet-applied results (member/updatedMember pairs, carrier types, target assignment overlays) to the next stage by hand, instead of simply reading already-applied state. Add a sealed CoordinatorOperationResult<T, U>, with CoordinatorResult as its terminal case (now final) and a new intermediate CoordinatorStep<T, U> (records + a continuation). CoordinatorWriteOperation.generateRecordsAndResult widens its return type to the sealed interface; every existing implementation already returns CoordinatorResult, so this is source-compatible and a no-behavior-change for every current coordinator. CoordinatorWriteEvent.run() replays and appends a CoordinatorStep's records as their own atomic chunk of the batch - through the existing append()/replay machinery, interleaved exactly as today - before invoking the continuation, so each step observes real, already-replayed state rather than a diff carried by hand. Only the terminal step's append attaches the write event, so the response future completes when the last step's records commit, even if earlier steps landed in an already-flushed batch. A replay failure on an intermediate step's append still fails the batch (the same as today), but additionally rethrows so the chain aborts and the write event completes exceptionally, since no event is attached to that append to do so otherwise. DeferredEventCollection.add() ignores a null event for the same reason. Transactional write operations reject a CoordinatorStep result: no transactional operation needs chaining, and prefix-commit semantics do not apply to them. No coordinator returns a CoordinatorStep yet, so this is a zero-behavior-change addition for group, share and transaction coordinators alike. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Building on the state-derived subscription-consistency check and the chained write-operation runtime support, decompose the regular consumer group heartbeat into three steps, each reading state that already includes the previous step's committed records instead of a hand-threaded diff: 1. consumerGroupHeartbeatUpdateMember - all validation, plus the member-state fence relocated from CurrentAssignmentBuilder (state known up front, and step 3 - where the builder runs - is a later durability unit); creates or updates the member and maybe bumps the group epoch. The epoch record is emitted atomically with the member record in this same step, which is why the what-if metadata-hash computation stays here. 2. maybeUpdateConsumerGroupTargetAssignment - reads group.members() and group.preferredServerAssignor() fresh (both already reflect step 1's replayed member), so the TargetAssignmentBuilder overlay (previously needed to see the not-yet-applied member) and the what-if computePreferredServerAssignor(old, new) are gone. 3. consumerGroupHeartbeatReconcile (terminal) - reconciles the member's current assignment (the only step that touches it) and builds the response. The leave paths and the classic-group-join/share/streams heartbeat paths are unchanged (own migration, per the design doc's follow-ups). Record content and order are preserved by construction, so GroupMetadataManagerTestContext mirrors the runtime by walking the chain and replaying each step before the next one runs, keeping existing assertions intact. One behavioral delta, matching prefix-commit semantics: on an assignor failure, step 1's member/epoch records now commit (previously nothing persisted), so a client retry converges without resending its subscription. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
dajac
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
WIP - Built with Claude to prototype the idea.
GroupMetadataManager.consumerGroupHeartbeat generates every record and
the response in one pass, before any of them are replayed to the state
machine. That forces every stage of the handler to hand-thread
not-yet-applied results to the next stage (records accumulators,
member/updatedMember pairs, carrier types, target-assignment overlays)
instead of just reading state — a bug-prone pattern where a forgotten
carrier read silently sees stale state, and a brand-new group's
soft-state writes are lost because the group isn't in the groups map
yet.
This change decomposes the heartbeat into steps that each observe state
already updated by the previous step's committed records, so staleness
becomes structurally impossible instead of a discipline problem.
Existing behavior is preserved by construction, with two small,
deliberate deltas (documented in the commits): pre-existing
subscription/assignment inconsistencies now self-heal instead of waiting
for an unrelated trigger, and an assignor failure now leaves a committed
member/epoch prefix that a retry converges on, instead of persisting
nothing.