KAFKA-20748: Keep StoredDescriptionTopologyEpoch consistent with the topology description plugin#22777
Open
aliehsaeedii wants to merge 7 commits into
Open
KAFKA-20748: Keep StoredDescriptionTopologyEpoch consistent with the topology description plugin#22777aliehsaeedii wants to merge 7 commits into
aliehsaeedii wants to merge 7 commits into
Conversation
…topology description plugin StoredDescriptionTopologyEpoch is the broker's record of which topology a Streams group's plugin entry holds, and it drives whether the broker solicits a fresh push, whether the entry still needs deleting, and what describe serves. The plugin call and the metadata write that records the epoch are non-atomic, so a crash or error between them, or a push that races a delete, could leave the broker's belief and the plugin permanently disagreeing: a leak (the plugin holds a topology the broker thinks absent, so nothing reclaims it) or a loss (the plugin holds nothing but the broker thinks an epoch is stored, so it never re-solicits and describe reports NOT_STORED forever). This makes the epoch three-valued by adding a -2 "uncertain" marker beside the existing real epoch (>= 0, definitely held) and -1 (definitely empty). Uncertain is read like -1 for the solicitation decision and like a real epoch for the deletion decision, so a group left uncertain both re-solicits and stays reclaimable. Every operation that disturbs the plugin commits -2 durably before the plugin call and writes the final value on success; any failure after the barrier leaves the group uncertain and therefore self-healing. The barrier is applied to the push path, the periodic expiration cleanup, DeleteGroups, and the conversion of an empty Streams group to a classic group on a classic JoinGroup, which previously tombstoned the Streams metadata and orphaned the plugin entry. The cleanup path also closes the delete-versus-push race: deleteTopology and a concurrent setTopology have no ordering, so after a successful delete the broker re-checks the stored epoch and clears it to -1 only if it is still uncertain, otherwise writes -2 back to force a re-solicit because the raced push may have been removed. The classic-JoinGroup conversion detects the empty-Streams case during the group lookup the join already performs, so non-Streams and already-classic joins add no extra operation on the hot path. Based on the draft by Lucas Brutschy (confluentinc/kafka@9f38e384ab), adapted to apache/kafka trunk. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Self-review fixes on the StoredDescriptionTopologyEpoch barrier design: - Honor the UNCERTAIN barrier write's response on the push path: when the group vanished between the validate read and the barrier write, no barrier record exists, so skip plugin.setTopology (which would create an entry no cleanup path ever reclaims) and fail the push with GROUP_ID_NOT_FOUND. - Re-arm UNCERTAIN in the post-push epoch write when the stored epoch is NONE: only a racing delete's finalize clears UNCERTAIN to NONE, so the pushed topology may have been wiped by plugin.deleteTopology; recording the pushed epoch would permanently suppress re-solicitation over an empty plugin. - Re-check group emptiness in markStoredDescriptionTopologyEpochUncertain for delete callers (markWhenNone=false): a group revived between the caller's committed read and the mark now drops out instead of having its plugin data deleted underneath the active member, making the "revived groups drop out" javadoc claim true. The classic-join conversion path now honors the mark's response accordingly and fails the join with a retriable error instead of running an unbarriered plugin delete. - Make the batched UNCERTAIN mark and post-delete finalize writes non-atomic, like the clearStoredDescriptionTopologyEpochBatch they replaced: the per-group records are independent, and an atomic batch exceeding the max batch size would permanently wedge the cleanup cycle for the partition. - Compare against STORED_TOPOLOGY_EPOCH_NONE instead of the literal -1 in StreamsGroup#shouldExpire and document that UNCERTAIN also defers the tombstone to keep the group reclaimable. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…oup ids - Short-circuit deleteStreamsTopologyDescriptions when the pre-delete read finds no streams groups with stored topology: the common DeleteGroups batch no longer schedules a mark-topology-uncertain-batch write that would append nothing and return an empty set. - Log the affected group ids (not just the count) when the post-delete finalize write fails, so an operator can tell which groups are stuck at UNCERTAIN(-2); the set is bounded by the per-partition eligibility scan. - Add a DeleteGroups test for the revived-group drop: when the mark batch returns an empty subset, plugin.deleteTopology is skipped, no plugin failure is recorded, and the pipeline still reaches the tombstone write (which reports NON_EMPTY_GROUP). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…rite failures - Arm a back-off window when the classic-join conversion delete fails and consult it before re-invoking the plugin: on REBALANCE_IN_PROGRESS the classic client retries the join immediately (no client-side back-off), so a broken plugin was hit with deleteTopology in a tight loop. The window is keyed at STORED_TOPOLOGY_EPOCH_UNCERTAIN — the group's stored epoch after the barrier write, never a real push epoch — so heartbeat/push windows are untouched, and it is dropped by the existing clearBackoffGroup on a successful conversion or cleanup-cycle delete. The conversion-cleanup chain moves into a cleanupTopologyBeforeConversion helper (joinGroup exceeded the NPath limit with the extra branch). - Swallow a failed UNCERTAIN mark write in the cleanup cycle with a partition-scoped warn naming the affected groups, mirroring the finalize write: a routine NOT_COORDINATOR during a shard move no longer fails the whole cycle's allOf with only the generic "failed to complete cleanly" log line. Skipping the shard's plugin delete is safe — the groups keep their stored epoch and the next cycle retries. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The delete-vs-push smart finalize only ran in the cleanup cycle, so a group revived between the UNCERTAIN barrier and the plugin delete could keep a raced push's epoch write (stored == UNCERTAIN -> max branch records the real epoch) over a plugin the delete just emptied: - DeleteGroups: finalize the groups whose plugin delete succeeded before the tombstone write, mirroring the cleanup cycle's mark -> delete -> finalize chain. A revived group survives its tombstone with NON_EMPTY_GROUP, and the finalize now clears or re-arms its stored epoch; for groups the tombstone removes the extra record is harmless. - Classic-join conversion: finalize after the re-join, where it no-ops for the converted (now classic) group and only writes for a group revived in between, whose re-join was rejected with INCONSISTENT_GROUP_PROTOCOL. - Document the residual window in the finalize javadoc: a raced push whose setTopology landed after the delete but reported a transient failure writes no epoch record, so clearing UNCERTAIN to NONE can leave plugin data at NONE until the back-off-solicited re-push heals it (or leaks it if the group is tombstoned first). - Drop two never() test assertions on the operation name is-empty-streams-group-with-stored-topology, which no production code uses, so they could never fail. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ion/finalize branches - listStreamsGroupsNeedingTopologyCleanup returns Set<String> instead of Map<String, Integer>: the observed stored epoch was dead (the mark batch re-checks the latest state itself; the cycle only uses the group ids), so the value no longer needs to be carried or documented. - Add GroupMetadataManager tests for the epoch-write and finalize branches the service-level runtime mocks never execute: the mainline success path (stored UNCERTAIN -> pushed epoch via max), the permanent-failure arm preserving the UNCERTAIN barrier, and the finalize no-op for a missing group and for a converted (non-streams) group. - Add a classicGroupJoin fall-through test under topologyCleanupHandled =false covering the non-eligible cases (missing, non-streams, non-empty streams, stored == NONE) so the detection predicate is pinned. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
DeleteGroups cleared the push-path back-off for every marked group, including those whose plugin.deleteTopology failed. A failed delete leaves the group at UNCERTAIN(-2), which re-solicits a push on the next heartbeat, so clearing the back-off let a rejoining member immediately re-attack the still-broken plugin. Clear only the succeeded groups (they are on their way to tombstone), matching the cleanup cycle's finalizeCleanupAfterDelete, and keep the failed groups' back-off so the interval-throttled cleanup cycle drives their retry instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
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
2 participants
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.
StoredDescriptionTopologyEpochis the broker's record of which topology aStreams group's plugin entry holds (KIP-1331), and it drives whether the broker
solicits a fresh push, whether the entry still needs deleting, and what
describe serves. The plugin call and the metadata write that records the epoch
are non-atomic, so a crash or error between them, or a push that races a
delete, could leave the broker's belief and the plugin permanently
disagreeing: a leak (the plugin holds a topology the broker thinks absent, so
nothing reclaims it) or a loss (the plugin holds nothing but the broker thinks
an epoch is stored, so it never re-solicits and describe reports NOT_STORED
forever).
This makes the epoch three-valued by adding a -2 "uncertain" marker beside the
existing real epoch (>= 0, definitely held) and -1 (definitely empty).
Uncertain is read like -1 for the solicitation decision and like a real epoch
for the deletion decision, so a group left uncertain both re-solicits and
stays reclaimable. Every operation that disturbs the plugin commits -2 durably
before the plugin call and writes the final value on success; any failure
after the barrier leaves the group uncertain and therefore self-healing. The
barrier is applied to the push path, the periodic expiration cleanup,
DeleteGroups, and the conversion of an empty Streams group to a classic group
on a classic JoinGroup, which previously tombstoned the Streams metadata and
orphaned the plugin entry.
The cleanup path also closes the delete-versus-push race:
deleteTopologyanda concurrent
setTopologyhave no ordering, so after a successful delete thebroker re-checks the stored epoch and clears it to -1 only if it is still
uncertain, otherwise writes -2 back to force a re-solicit because the raced
push may have been removed. The classic-JoinGroup conversion detects the
empty-Streams case during the group lookup the join already performs, so
non-Streams and already-classic joins add no extra operation on the hot path.
Testing
The full
:group-coordinatortest suite passes (1,982 tests, 0 failures),including new tests covering the mark-then-delete-then-finalize ordering of
the cleanup cycle, the revived-group drop-out on the barrier write, the
uncertain-state solicitation and deletion decisions, and the deferred
classic-JoinGroup conversion.
JIRA: https://issues.apache.org/jira/browse/KAFKA-20748
🤖 Generated with Claude Code