squah-confluent commented on code in PR #23514:
URL: https://github.com/apache/kafka/pull/23514#discussion_r4306038980
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -24457,6 +24457,131 @@ public void testStreamsGroupEvaluatedConfigs() {
result.response().data().taskOffsetIntervalMs());
}
+ private static Stream<CoordinatorRecord> consumerGroupReplayValues() {
+ String groupId = "group-id";
+ ConsumerGroupMember member = new
ConsumerGroupMember.Builder("member-id")
+ .setSubscribedTopicNames(List.of("foo"))
+ .setMemberEpoch(10)
+ .build();
+ return Stream.of(
+
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId,
member),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId,
10, 0),
+ CoordinatorRecord.record(
+ new ConsumerGroupPartitionMetadataKey().setGroupId(groupId),
+ new ApiMessageAndVersion(new
ConsumerGroupPartitionMetadataValue(), (short) 0)
+ ),
+
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId,
"member-id", Map.of()),
+
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId,
10, 12345L),
+
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId,
member),
+
GroupCoordinatorRecordHelpers.newConsumerGroupRegularExpressionRecord(
+ groupId, "foo.*", new ResolvedRegularExpression(Set.of("foo"),
1, 12345L))
+ );
+ }
+
+ private static Stream<CoordinatorRecord> consumerGroupReplayTombstones() {
+ return consumerGroupReplayValues().map(record ->
CoordinatorRecord.tombstone(record.key()));
+ }
+
+ @ParameterizedTest
+ @MethodSource("consumerGroupReplayValues")
+ public void
testReplayConsumerGroupValueAfterClassicGroup(CoordinatorRecord record) {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord("group-id",
+ new
GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+ ClassicGroup classicGroup =
context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false);
+ assertFalse(classicGroup.isSimpleGroup());
+ context.commit();
+
+ // The loader may have read the classic record before compaction
removed the
+ // upgrade tombstone. Each consumer value must be able to recreate the
group.
+ context.replay(record);
+ assertEquals(Group.GroupType.CONSUMER,
context.groupMetadataManager.group("group-id").type());
+
+ // Replacing the group during replay must also respect snapshot
rollback.
+ context.rollback();
+ assertSame(classicGroup,
context.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("consumerGroupReplayTombstones")
+ public void
testReplayConsumerGroupTombstoneAfterClassicGroup(CoordinatorRecord record) {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord("group-id",
+ new
GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+ ClassicGroup classicGroup =
context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false);
+
+ context.replay(record);
+ assertSame(classicGroup,
context.groupMetadataManager.group("group-id"));
+ assertEquals(1, classicGroup.generationId());
+ }
+
+ @ParameterizedTest
+ @MethodSource("consumerGroupReplayTombstones")
+ public void
testReplayConsumerGroupTombstonePreservesSimpleClassicGroup(CoordinatorRecord
record) {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+ ClassicGroup classicGroup =
context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", true);
+ assertTrue(classicGroup.isSimpleGroup());
+
+ context.replay(record);
+ assertSame(classicGroup,
context.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @MethodSource({"consumerGroupReplayValues",
"consumerGroupReplayTombstones"})
+ public void
testReplayConsumerGroupRecordRejectsUnrelatedGroupTypes(CoordinatorRecord
record) {
Review Comment:
A few things here:
1. #23530 and KAFKA-21102 handle missing tombstones within a single group
type. Neither changes how records of one type are replayed against a group of
another type, which is what this PR is about.
2. This isn't specific to consumer groups or downgrades. Deleting and
recreating a group with a different type is enough, for any pair of types:
```
1 T1 records for G G created as T1
2 T1 tombstones for G G deleted <- group tombstone compacted,
superseded by step 5
3 T2 records for G G recreated as T2
4 T2 tombstones for G G deleted
5 T1 records for G G recreated as T1
```
A load that reads step 1, then a compacted step 2, then step 3 replays T2
records against a T1 group. Analogous to #23530, where the final group
tombstone implies the rest of the group's tombstones, I think the simplest
consistent rule is: any record of one group type replayed against a group of
another type removes the existing group, after which a value creates the new
group as usual and a tombstone is a no-op. That should apply to all group types
(e.g. `getOrMaybeCreatePersistedStreamsGroup` has the same check), and it
subsumes the `isSimpleGroup()` special cases.
3. The removal must go through `removeGroup`, not just `groups.put` of the
new group. Otherwise the replaced group's bookkeeping, such as its
`groupsByTopics` subscriptions, is left behind. The cleanup in `removeGroup`
comes from #23530, so this depends on it.
4. Classic `GroupMetadata` value replay has the same problem today: it does
`groups.put` over whatever group exists without calling `removeGroup`.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -24457,6 +24457,131 @@ public void testStreamsGroupEvaluatedConfigs() {
result.response().data().taskOffsetIntervalMs());
}
+ private static Stream<CoordinatorRecord> consumerGroupReplayValues() {
+ String groupId = "group-id";
+ ConsumerGroupMember member = new
ConsumerGroupMember.Builder("member-id")
+ .setSubscribedTopicNames(List.of("foo"))
+ .setMemberEpoch(10)
+ .build();
+ return Stream.of(
+
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId,
member),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId,
10, 0),
+ CoordinatorRecord.record(
+ new ConsumerGroupPartitionMetadataKey().setGroupId(groupId),
+ new ApiMessageAndVersion(new
ConsumerGroupPartitionMetadataValue(), (short) 0)
+ ),
+
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId,
"member-id", Map.of()),
+
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId,
10, 12345L),
+
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId,
member),
+
GroupCoordinatorRecordHelpers.newConsumerGroupRegularExpressionRecord(
+ groupId, "foo.*", new ResolvedRegularExpression(Set.of("foo"),
1, 12345L))
+ );
+ }
+
+ private static Stream<CoordinatorRecord> consumerGroupReplayTombstones() {
+ return consumerGroupReplayValues().map(record ->
CoordinatorRecord.tombstone(record.key()));
+ }
+
+ @ParameterizedTest
+ @MethodSource("consumerGroupReplayValues")
+ public void
testReplayConsumerGroupValueAfterClassicGroup(CoordinatorRecord record) {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord("group-id",
+ new
GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+ ClassicGroup classicGroup =
context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false);
+ assertFalse(classicGroup.isSimpleGroup());
+ context.commit();
+
+ // The loader may have read the classic record before compaction
removed the
+ // upgrade tombstone. Each consumer value must be able to recreate the
group.
+ context.replay(record);
+ assertEquals(Group.GroupType.CONSUMER,
context.groupMetadataManager.group("group-id").type());
+
+ // Replacing the group during replay must also respect snapshot
rollback.
+ context.rollback();
+ assertSame(classicGroup,
context.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("consumerGroupReplayTombstones")
+ public void
testReplayConsumerGroupTombstoneAfterClassicGroup(CoordinatorRecord record) {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+
context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord("group-id",
+ new
GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+ ClassicGroup classicGroup =
context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false);
+
+ context.replay(record);
+ assertSame(classicGroup,
context.groupMetadataManager.group("group-id"));
+ assertEquals(1, classicGroup.generationId());
+ }
+
+ @ParameterizedTest
+ @MethodSource("consumerGroupReplayTombstones")
+ public void
testReplayConsumerGroupTombstonePreservesSimpleClassicGroup(CoordinatorRecord
record) {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder().build();
+ ClassicGroup classicGroup =
context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", true);
Review Comment:
Simple groups do need their own coverage: the new tombstone branch handles
them incorrectly. Simple groups are created by offset commit replay, so they
can be observed without the log ever containing a classic group:
```
1 ConsumerGroup* values for G <- compacted
2 OffsetCommit(G, foo-0) <- replay creates a simple classic group
3 DeleteGroups: offset tombstone + ConsumerGroup* tombstones for G
```
Replaying the uncompacted log leaves no group. With this change, an empty
simple group is left behind, since the consumer tombstones are ignored and
offset tombstones don't remove groups. Under the rule above, the consumer
tombstones remove it. The removed
`testReplayConsumerGroupTombstonePreservesSimpleClassicGroup` covered this case
but asserted the wrong behavior.
--
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]
