MINOR: Add compaction replay tests for the group coordinator - #23220
Open
izzyharker wants to merge 12 commits into
Open
MINOR: Add compaction replay tests for the group coordinator#23220izzyharker wants to merge 12 commits into
izzyharker wants to merge 12 commits into
Conversation
Adds GroupMetadataManagerCompactionReplayTest, which catches "compaction
makes a partition fail to load" bugs by replaying every realistic compacted
variant of a real group-coordinator record log through a fresh coordinator.
For each scenario it captures the records the coordinator writes (one batch
per write), then sweeps the clean/dirty compaction boundary across every batch
boundary, with tombstones retained and dropped, replaying each variant and
asserting it loads without throwing. Offset-commit records are replayed through
an OffsetMetadataManager sharing the GroupMetadataManager, so the
simple-classic-group creation on the load path is modelled faithfully.
The clean/dirty boundary is only split between batches, never inside one,
because log compaction cleans at segment granularity and a single write's
records are always compacted atomically. Splitting within a batch generates
logs real compaction can never produce (false positives).
Scenarios: rebalance, subscription change, member leave (tombstones), static
member rejoin, classic->consumer upgrade, and consumer/streams groups with an
offset commit. The last two reproduce a bug where, after compaction, the
offset commit sorts before the group records, creating a simple classic group
that the consumer/streams records must load on top of. Both fail if their
respective fixes in getOrMaybeCreatePersisted{Consumer,Streams}Group are
reverted.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… isolation Restore the fresh-coordinator-per-variant isolation in assertCompactedVariantsLoadCleanly (a Supplier, not a single reused context) so compacted variants can no longer leak state into one another. Add two scenarios that reproduce the simple-classic-group load path over a group that started classic and holds committed offsets: - testClassicToConsumerUpgradeWithOffsetCommit... (upgrade to consumer) - testClassicToStreamsMigrationWithOffsetCommit... (offline migration to streams) Both were verified to fail (IllegalStateException "not a consumer/streams group") when the respective isSimpleGroup() branch is disabled. Make the docstrings consistent: tag the streams paths to KAFKA-20254, describe the consumer paths as its pre-existing counterpart, and scope out the "still owned at epoch" path (covered by GroupMetadataManagerTest) since the clean-prefix/dirty-suffix model cannot reproduce it. Fix a typo and trailing whitespace. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Replace the per-batch clean-prefix model with a single cleaning window over
the log: a record is compactable when it is superseded by a later record with
the same key or is a tombstone, and each variant removes every compactable
record in one contiguous stretch. This subsumes the previous prefix-only model
and, because the window can start in the middle of the log, it also produces
the missed-unassignment shape behind KAFKA-19862, which a compacted prefix
cannot express. The separate drop-tombstones flag is gone.
Two constraints keep the variants to logs a load can actually observe: the
window falls on batch boundaries, since the records of one write share a
segment and therefore share the cleaner's verdict; and a tombstone is only
dropped once no earlier record for its key survives, since dropping an
aged-out tombstone happens in a later pass than the one that collapsed the key.
Collapse the nine narrow scenarios into two lifecycle scenarios driven through
the real request paths, as reviewed: a group created on the classic protocol
that commits offsets, rebalances, is upgraded online to the consumer protocol,
rolls its classic members onto the new protocol and scales out; and a Kafka
Streams application migrated offline from the classic protocol to the streams
protocol across three processes. Both mix classic GroupMetadata records and
their tombstone, offset commits and modern group records, and move partitions
or tasks between owners repeatedly.
Verified by reverting each fix in turn and re-running:
- Streams simple-classic-group handling in
getOrMaybeCreatePersistedStreamsGroup: "Group ... is not a streams group"
- KAFKA-19862 ConsumerGroup partition epochs: "Cannot set the epoch of ...
because the partition is still owned at epoch ..."
- KAFKA-19862 StreamsGroup process ids: "Cannot remove the process ID ...
because it does not have any processId"
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Reuse of the existing harness: - StreamsGroupTestUtil.staticHeartbeat/staticJoinHeartbeat now build the streams join, leave and reconcile requests. - Remove follow SyncGroup block, default to SyncGroupRequestBuilder Deduplication: - Add leaveStreamsMember, joinConsumerMember, joinStreamsMember and sleepCapturing, and fold the two reconciliation loops into a single driver. Compaction model: - compact() returns a CompactedVariant carrying the records, the cleaned window and the removed positions. - Use a single GroupCoordinatorMetrics instance. - Pass withLogContext to the OffsetMetadataManager.Builder, as production does. The captured logs are unchanged: both scenarios still produce exactly the same records, batches and compactable positions as before (69/30/52 for the consumer scenario, 51/26/42 for the streams one). Verified by reverting each fix and re-running: KAFKA-20254 fails the streams scenario, KAFKA-19862 fails both. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
github-actions
Bot
added
triage
squah-confluent
left a comment
squah-confluent
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the tests!
| * commits and member joins/rebalances for an online classic -> consumer upgrade and an | ||
| * offline classic -> streams upgrade. Both tests capture written records, compact the | ||
| * resulting log, and replay records through a new group coordinator shard to verify loading. | ||
| * Multiple contiguous log segments are tested, including prefix and mid-log windows. |
Contributor
There was a problem hiding this comment.
This could use a little more explanation. We're testing two compaction cases here:
- The typical compaction everyone thinks of, where a prefix of the log is compacted.
- Compaction concurrent with a load, where the load reads an uncompacted section of the log, compaction races ahead, the load reads a compacted section, then reaches the uncompacted active segment. This is the cause of KAFKA-19862
We really need to explain how the second case arises, otherwise it's not clear to the reader why we're testing mid-log compaction. Including the ticket number lets the reader know it actually bites in production and is not some contrived case.
| * offline classic -> streams upgrade. Both tests capture written records, compact the | ||
| * resulting log, and replay records through a new group coordinator shard to verify loading. | ||
| * Multiple contiguous log segments are tested, including prefix and mid-log windows. | ||
| * |
Contributor
There was a problem hiding this comment.
This is also a good place to explain why we align compaction to batch boundaries instead of simplifying the model.
ConsumerGroupMemberMetadataKey
...
ConsumerGroupCurrentMemberAssignmentKey, compacted
--- batch boundary ---
ConsumerGroupCurrentMemberAssignmentKey = tombstone, compacted
ConsumerGroupTargetAssignmentMemberKey = tombstone
ConsumerGroupMemberMetadataKey = tombstone <-- fails
ConsumerGroupMemberMetadataKey initializes the member epoch to 0, while the ConsumerGroupCurrentMemberAssignmentKey tombstone updates the member epoch to LEAVE_GROUP_MEMBER_EPOCH. The ConsumerGroupMemberMetadataKey tombstone expects a member epoch of LEAVE_GROUP_MEMBER_EPOCH and not 0, so replay fails when the ConsumerGroupCurrentMemberAssignmentKey tombstone is compacted.
(but in better sentences)
Normally I'd be against putting this kind of low-level detail in a class-level javadoc but I can't find another place to put it.
| private CoordinatorMetadataImage metadataImage; | ||
| private MockPartitionAssignor consumerAssignor; | ||
| private MockTaskAssignor streamsAssignor; | ||
| private GroupMetadataManagerTestContext context; |
Contributor
There was a problem hiding this comment.
I'm uncomfortable with putting mutable state in an instance variable shared by all test methods. Technically it's fine since the test runner ought to guarantee that the test methods aren't run concurrently.
| // Verify the partitions can be reloaded cleanly from log. | ||
| assertCompactedVariantsLoadCleanly(capturedLog); | ||
| } | ||
|
|
Contributor
There was a problem hiding this comment.
How much extra work would it be to add downgrade tests (downgrade by leave, downgrade by static member replacement and streams offline downgrade)?
Contributor
Author
There was a problem hiding this comment.
I had Claude give this a shot and the first two had no issues (leave/static member replacement) but it did flag a concurrent compaction-related bug with the streams offline downgrade test (similar IllegalStateException case as KAFKA-19862).
Something like the following can happen with concurrent compaction (here the compaction window is records 24-28) when a streams group gets downgraded offline:
10 | StreamsGroupMetadataKey(...) <- not compacted
...
24 | StreamsGroupTargetAssignmentMetadataKey(...) [compacted]
25 | StreamsGroupTargetAssignmentMetadataKey(...) = tombstone [compacted] <- delete tombstone removed
26 | StreamsGroupMetadataKey(...) = tombstone <-- replay failed (group delete)
27 | StreamsGroupTopologyKey(...) = tombstone
28 | GroupMetadataKey(...) [compacted]
29 | GroupMetadataKey(...) <- new classic group
...
ERROR: java.lang.IllegalStateException: Received a tombstone record to delete group <id>
but did not receive StreamsGroupTargetAssignmentMetadataValue tombstone.
at GroupMetadataManager.replay(GroupMetadataManager.java:6264)
Because the initial StreamsGroupMetadata log is outside the compaction window, the StreamsGroupTargetAssignmentMetadata records are compacted away but the later StreamsGroupMetadata tombstone remains.
Contributor
Author
There was a problem hiding this comment.
The failing test is in this commit -> confluentinc@99c40a3
- Added explanation to initial javadoc for test scenarios and compaction - Small syntax/etc updates
Move the request/scenario helpers and the record accumulator out of the test class into a dedicated CompactionReplayTestContext, which owns the captured log internally (inlining the former CapturedLog) so scenarios no longer thread a log through every call. The test retains the compaction model and load assertions. Also drop the banner-style section comments. Capture records for the member-B classic joins via joinClassicMember, for consistency with the member-A rejoin. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Split assertCompactedVariantsLoadCleanly into its three cases with signposting comments: uncompacted, compacted prefix (the standard case, previously buried in the general window loop), and concurrent compaction. Drop the CompactedVariant record. assertLoadsCleanly now takes the full log plus the set of compacted positions, and on failure renders the whole log one record per line with position, marking tombstones, compacted records, and the record that failed to replay. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add an append overload that inspects CoordinatorResult.isAtomic(): an atomic result is recorded as a single batch (compaction cannot split it), while a non-atomic result is recorded as one batch per record, letting compaction fall between any two of its records. Route the coordinator-result call sites (heartbeats, classic leave, timeouts) through it. Every operation these tests exercise is atomic today, so the recorded batches are unchanged; this future-proofs the model against a coordinator op becoming non-atomic. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add two downgrade scenarios that mirror the existing upgrade coverage: - testConsumerGroupDowngradeByLeave: a classic group is upgraded online to a consumer group, then the last consumer-protocol member leaves, downgrading the group back to classic. - testConsumerGroupDowngradeByStaticMemberReplacement: a classic member replaces the last consumer-protocol static member, downgrading the group. Both assert the group type before and after to confirm the downgrade fired. Switch the migration policy to BIDIRECTIONAL (the production default) so downgrades are allowed; the existing upgrade scenarios are unchanged. Add leaveConsumerMember, joinStaticConsumerMember, replaceStaticMemberWithClassicProtocol and groupType helpers to the test context. Co-Authored-By: Claude Opus 4.8 (1M context) <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
3 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.
Adds GroupMetadataManagerCompactionReplayTest, which tests partition
loading after compaction for scenarios involving offset commit combined
with classic->consumer/streams upgrades, as seen in
KAFKA-19862/KAFKA-20254.
Tests capture records, compact a contiguous stretch of the resulting log
(both prefix and mid-log), and replay the result(s) through a new
GroupCoordinatorShard.
testClassicGroupUpgradeToConsumerGroupWithOffsetCommit: A classicgroup has an offset commit, rebalance, consumer member join, online
upgrade (classic->consumer), new member join. Verified this test fails
when KAFKA-19862 is reverted.
testClassicGroupUpgradeToStreamsGroupWithOffsetCommit: Streams appruns on classic protocol, commits offsets, rebalances, then upgrades
offline to streams protocol. Verified this test fails when KAFKA-20254
is reverted.
Reviewers: Sean Quah squah@confluent.io