You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gitedmond (via GitHub)" <gi...@apache.org> on 2026/09/18 18:06:44 UTC
[PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
gitedmond opened a new pull request, #23514:
URL: https://github.com/apache/kafka/pull/23514
Kafka can fail to load a group during an online consumer-to-classic downgrade when log compaction races with coordinator replay. The loader may encounter a classic group followed by consumer-group records whose upgrade tombstone was removed, and throws `Group <id> is not a consumer group`. Consumer tombstones from the downgrade can also arrive after the classic group is loaded.
This change lets replay create a consumer group from a positive consumer record when the group currently has the classic type, and treats consumer tombstones against a classic group as no-ops. Validation for Streams and Share group collisions remains strict.
The PR adds record-level regression tests covering migration, tombstones, unrelated group types, and snapshot rollback. It also enables the two consumer downgrade compaction-replay tests added in #23220. Both of those tests failed with the reported exception before this change and pass afterward.
Validation: `:group-coordinator:test :group-coordinator:spotlessCheck` (including Checkstyle and SpotBugs) on current trunk. The focused before/after reproduction was verified separately against the unmodified replay helper.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23514:
URL: https://github.com/apache/kafka/pull/23514#discussion_r4157967387
##########
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) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.replay(GroupCoordinatorRecordHelpers.newShareGroupEpochRecord("group-id", 10, 0));
+ Group shareGroup = context.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> context.replay(record));
+ assertSame(shareGroup, context.groupMetadataManager.group("group-id"));
+
+ GroupMetadataManagerTestContext streamsContext = new GroupMetadataManagerTestContext.Builder().build();
+ streamsContext.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ "group-id", 10, 0, 0, getDefaultAssignmentConfigs(), -1, -1));
+ Group streamsGroup = streamsContext.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> streamsContext.replay(record));
+ assertSame(streamsGroup, streamsContext.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplayConsumerGroupMigrationWithCompactedUpgradeTombstone(boolean compacted) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ String groupId = "group-id";
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+
+ // A concurrent downgrade can let compaction remove the upgrade tombstone
+ // after the loader has already read the older classic metadata record.
+ if (!compacted) {
+ context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+ }
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+ assertEquals(10, consumerGroup.groupEpoch());
+ assertEquals(10, consumerGroup.getOrMaybeCreateMember("member-id", false).memberEpoch());
+
+ List<CoordinatorRecord> tombstones = new ArrayList<>();
+ consumerGroup.createGroupTombstoneRecords(tombstones);
+ tombstones.forEach(context::replay);
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(2)));
+ assertEquals(2, context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false).generationId());
+ }
+
+ @Test
+ public void testReplayConsumerGroupTombstoneStillRejectsLiveMembers() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup group = context.groupMetadataManager.consumerGroup("group-id");
+
+ IllegalStateException exception = assertThrows(IllegalStateException.class, () -> context.replay(
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochTombstoneRecord("group-id")));
+ assertEquals("Received a tombstone record to delete group group-id but the group still has 1 members.",
+ exception.getMessage());
+ assertSame(group, context.groupMetadataManager.group("group-id"));
+ }
Review Comment:
This is handled by #23530
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Handle group type changes during replay [kafka]
Posted by "gitedmond (via GitHub)" <gi...@apache.org>.
gitedmond commented on PR #23514:
URL: https://github.com/apache/kafka/pull/23514#issuecomment-6253356749
Thanks, Sean. You were right about cross-type replay and simple-group tombstones. The updated patch consistently removes mismatched groups through `removeGroup`, using #23530’s cleanup, before creating a new group for values or leaving it absent for tombstones.
I also reorganized the parameterized tests as requested, added bookkeeping cleanup assertions and the simple-group offset-replay regression, and retained the downgrade compaction coverage.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "gitedmond (via GitHub)" <gi...@apache.org>.
gitedmond commented on code in PR #23514:
URL: https://github.com/apache/kafka/pull/23514#discussion_r4158386982
##########
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) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.replay(GroupCoordinatorRecordHelpers.newShareGroupEpochRecord("group-id", 10, 0));
+ Group shareGroup = context.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> context.replay(record));
+ assertSame(shareGroup, context.groupMetadataManager.group("group-id"));
+
+ GroupMetadataManagerTestContext streamsContext = new GroupMetadataManagerTestContext.Builder().build();
+ streamsContext.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ "group-id", 10, 0, 0, getDefaultAssignmentConfigs(), -1, -1));
+ Group streamsGroup = streamsContext.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> streamsContext.replay(record));
+ assertSame(streamsGroup, streamsContext.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplayConsumerGroupMigrationWithCompactedUpgradeTombstone(boolean compacted) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ String groupId = "group-id";
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+
+ // A concurrent downgrade can let compaction remove the upgrade tombstone
+ // after the loader has already read the older classic metadata record.
+ if (!compacted) {
+ context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+ }
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+ assertEquals(10, consumerGroup.groupEpoch());
+ assertEquals(10, consumerGroup.getOrMaybeCreateMember("member-id", false).memberEpoch());
+
+ List<CoordinatorRecord> tombstones = new ArrayList<>();
+ consumerGroup.createGroupTombstoneRecords(tombstones);
+ tombstones.forEach(context::replay);
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(2)));
+ assertEquals(2, context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false).generationId());
+ }
+
+ @Test
+ public void testReplayConsumerGroupTombstoneStillRejectsLiveMembers() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup group = context.groupMetadataManager.consumerGroup("group-id");
+
+ IllegalStateException exception = assertThrows(IllegalStateException.class, () -> context.replay(
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochTombstoneRecord("group-id")));
+ assertEquals("Received a tombstone record to delete group group-id but the group still has 1 members.",
+ exception.getMessage());
+ assertSame(group, context.groupMetadataManager.group("group-id"));
+ }
Review Comment:
Removed this test in 60d36392a0. PR #23530 changes the expected group metadata tombstone behavior and tests cleanup when sibling tombstones are missing, so this PR no longer asserts the old exception.
##########
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) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.replay(GroupCoordinatorRecordHelpers.newShareGroupEpochRecord("group-id", 10, 0));
+ Group shareGroup = context.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> context.replay(record));
+ assertSame(shareGroup, context.groupMetadataManager.group("group-id"));
+
+ GroupMetadataManagerTestContext streamsContext = new GroupMetadataManagerTestContext.Builder().build();
+ streamsContext.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ "group-id", 10, 0, 0, getDefaultAssignmentConfigs(), -1, -1));
+ Group streamsGroup = streamsContext.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> streamsContext.replay(record));
+ assertSame(streamsGroup, streamsContext.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplayConsumerGroupMigrationWithCompactedUpgradeTombstone(boolean compacted) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ String groupId = "group-id";
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+
+ // A concurrent downgrade can let compaction remove the upgrade tombstone
+ // after the loader has already read the older classic metadata record.
+ if (!compacted) {
+ context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+ }
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+ assertEquals(10, consumerGroup.groupEpoch());
+ assertEquals(10, consumerGroup.getOrMaybeCreateMember("member-id", false).memberEpoch());
+
+ List<CoordinatorRecord> tombstones = new ArrayList<>();
+ consumerGroup.createGroupTombstoneRecords(tombstones);
+ tombstones.forEach(context::replay);
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(2)));
+ assertEquals(2, context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false).generationId());
+ }
Review Comment:
Agreed; I removed the duplicate migration sequence test in 60d36392a0. The parameterized value test still checks every consumer record and snapshot rollback, while the two enabled compaction replay tests cover the full downgrade sequences.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23514:
URL: https://github.com/apache/kafka/pull/23514#discussion_r4157985338
##########
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) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.replay(GroupCoordinatorRecordHelpers.newShareGroupEpochRecord("group-id", 10, 0));
+ Group shareGroup = context.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> context.replay(record));
+ assertSame(shareGroup, context.groupMetadataManager.group("group-id"));
+
+ GroupMetadataManagerTestContext streamsContext = new GroupMetadataManagerTestContext.Builder().build();
+ streamsContext.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ "group-id", 10, 0, 0, getDefaultAssignmentConfigs(), -1, -1));
+ Group streamsGroup = streamsContext.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> streamsContext.replay(record));
+ assertSame(streamsGroup, streamsContext.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplayConsumerGroupMigrationWithCompactedUpgradeTombstone(boolean compacted) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ String groupId = "group-id";
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+
+ // A concurrent downgrade can let compaction remove the upgrade tombstone
+ // after the loader has already read the older classic metadata record.
+ if (!compacted) {
+ context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+ }
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+ assertEquals(10, consumerGroup.groupEpoch());
+ assertEquals(10, consumerGroup.getOrMaybeCreateMember("member-id", false).memberEpoch());
+
+ List<CoordinatorRecord> tombstones = new ArrayList<>();
+ consumerGroup.createGroupTombstoneRecords(tombstones);
+ tombstones.forEach(context::replay);
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(2)));
+ assertEquals(2, context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false).generationId());
+ }
Review Comment:
Isn't this test the same as `testReplayConsumerGroupValueAfterClassicGroup`?
##########
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) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.replay(GroupCoordinatorRecordHelpers.newShareGroupEpochRecord("group-id", 10, 0));
+ Group shareGroup = context.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> context.replay(record));
+ assertSame(shareGroup, context.groupMetadataManager.group("group-id"));
+
+ GroupMetadataManagerTestContext streamsContext = new GroupMetadataManagerTestContext.Builder().build();
+ streamsContext.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ "group-id", 10, 0, 0, getDefaultAssignmentConfigs(), -1, -1));
+ Group streamsGroup = streamsContext.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> streamsContext.replay(record));
+ assertSame(streamsGroup, streamsContext.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplayConsumerGroupMigrationWithCompactedUpgradeTombstone(boolean compacted) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ String groupId = "group-id";
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+
+ // A concurrent downgrade can let compaction remove the upgrade tombstone
+ // after the loader has already read the older classic metadata record.
+ if (!compacted) {
+ context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+ }
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+ assertEquals(10, consumerGroup.groupEpoch());
+ assertEquals(10, consumerGroup.getOrMaybeCreateMember("member-id", false).memberEpoch());
+
+ List<CoordinatorRecord> tombstones = new ArrayList<>();
+ consumerGroup.createGroupTombstoneRecords(tombstones);
+ tombstones.forEach(context::replay);
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(2)));
+ assertEquals(2, context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false).generationId());
+ }
+
+ @Test
+ public void testReplayConsumerGroupTombstoneStillRejectsLiveMembers() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup group = context.groupMetadataManager.consumerGroup("group-id");
+
+ IllegalStateException exception = assertThrows(IllegalStateException.class, () -> context.replay(
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochTombstoneRecord("group-id")));
+ assertEquals("Received a tombstone record to delete group group-id but the group still has 1 members.",
+ exception.getMessage());
+ assertSame(group, context.groupMetadataManager.group("group-id"));
+ }
Review Comment:
This is handled by #23530 and the intended behavior will change.
##########
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:
In the replay tests, we usually perform test setup by replaying records.
##########
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:
Should we reject unrelated group types? Does the bug generalize to other group type changes?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23514:
URL: https://github.com/apache/kafka/pull/23514#discussion_r4157967387
##########
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) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.replay(GroupCoordinatorRecordHelpers.newShareGroupEpochRecord("group-id", 10, 0));
+ Group shareGroup = context.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> context.replay(record));
+ assertSame(shareGroup, context.groupMetadataManager.group("group-id"));
+
+ GroupMetadataManagerTestContext streamsContext = new GroupMetadataManagerTestContext.Builder().build();
+ streamsContext.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ "group-id", 10, 0, 0, getDefaultAssignmentConfigs(), -1, -1));
+ Group streamsGroup = streamsContext.groupMetadataManager.group("group-id");
+ assertThrows(IllegalStateException.class, () -> streamsContext.replay(record));
+ assertSame(streamsGroup, streamsContext.groupMetadataManager.group("group-id"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplayConsumerGroupMigrationWithCompactedUpgradeTombstone(boolean compacted) {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ String groupId = "group-id";
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(1)));
+
+ // A concurrent downgrade can let compaction remove the upgrade tombstone
+ // after the loader has already read the older classic metadata record.
+ if (!compacted) {
+ context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+ }
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup consumerGroup = context.groupMetadataManager.consumerGroup(groupId);
+ assertEquals(10, consumerGroup.groupEpoch());
+ assertEquals(10, consumerGroup.getOrMaybeCreateMember("member-id", false).memberEpoch());
+
+ List<CoordinatorRecord> tombstones = new ArrayList<>();
+ consumerGroup.createGroupTombstoneRecords(tombstones);
+ tombstones.forEach(context::replay);
+ context.replay(GroupMetadataManagerTestContext.newGroupMetadataRecord(groupId,
+ new GroupMetadataValue().setProtocolType("consumer").setGeneration(2)));
+ assertEquals(2, context.groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, false).generationId());
+ }
+
+ @Test
+ public void testReplayConsumerGroupTombstoneStillRejectsLiveMembers() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ consumerGroupReplayValues().forEach(context::replay);
+ ConsumerGroup group = context.groupMetadataManager.consumerGroup("group-id");
+
+ IllegalStateException exception = assertThrows(IllegalStateException.class, () -> context.replay(
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochTombstoneRecord("group-id")));
+ assertEquals("Received a tombstone record to delete group group-id but the group still has 1 members.",
+ exception.getMessage());
+ assertSame(group, context.groupMetadataManager.group("group-id"));
+ }
Review Comment:
This is handled by #23530
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #23514:
URL: https://github.com/apache/kafka/pull/23514#issuecomment-5883005711
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21117: Allow consumer group replay across classic downgrade [kafka]
Posted by "gitedmond (via GitHub)" <gi...@apache.org>.
gitedmond commented on code in PR #23514:
URL: https://github.com/apache/kafka/pull/23514#discussion_r4158389411
##########
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:
I removed the new Share/Streams rejection test in 60d36392a0 so this PR does not establish that behavior as a new contract. The production change remains limited to replaying consumer records after a classic group. A similar race may be possible for other transitions, but I have not reproduced a Share/Streams collision; KAFKA-21102 and #23530 address broader tombstone replay. Changing another type transition should follow a concrete compaction replay case.
##########
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:
Removed the simple classic tombstone test that constructed its setup directly. The retained non-simple classic tombstone test creates the group by replaying a classic metadata record, then checks that each consumer tombstone leaves it intact. Both classic variants take the same new tombstone branch.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org