squah-confluent commented on code in PR #23514:
URL: https://github.com/apache/kafka/pull/23514#discussion_r4306039008
##########
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:
That branch is the problem for simple groups. Simple groups are created by
offset commit replay, so they can be observed without the log ever containing a
classic group:
```
o1 ConsumerGroup* values for G <- compacted
o2 OffsetCommit(G, foo-0) <- replay creates a simple classic group
o3 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` asserted the
opposite.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -24457,6 +24457,64 @@ public void testStreamsGroupEvaluatedConfigs() {
result.response().data().taskOffsetIntervalMs());
}
+ private static Stream<CoordinatorRecord> consumerGroupReplayValues() {
Review Comment:
On the new tests:
- Please restructure them into
`testReplay(Consumer|Streams|Share)GroupRecord(Tombstone)?WithOtherGroupType`,
parameterized over (previous group type, record type), and
`testReplayGroupMetadata(Tombstone)?WithOtherGroupType`, parameterized over the
previous group type. Previous group types should include classic, simple
classic and the other modern types, and the tests should assert the previous
group's bookkeeping is cleaned up, using the same approach as #23530.
- Please place each at the end of the existing replay tests for its group
type, e.g. the consumer ones after
`testReplayConsumerGroupCurrentMemberAssignmentTombstone`.
##########
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:
```
o1 T1 records for G G created as T1
o2 T1 tombstones for G G deleted <- group tombstone compacted,
superseded by o5
o3 T2 records for G G recreated as T2
o4 T2 tombstones for G G deleted
o5 T1 records for G G recreated as T1
```
A load that reads o1, then a compacted o2, then o3 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`.
--
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]
