izzyharker commented on code in PR #23220: URL: https://github.com/apache/kafka/pull/23220#discussion_r4039641349
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardCompactionReplayTest.java: ########## @@ -0,0 +1,674 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.JoinGroupResponseData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.record.internal.RecordBatch; +import org.apache.kafka.common.utils.internals.LogContext; +import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage; +import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord; +import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder; +import org.apache.kafka.coordinator.group.CompactionReplayTestContext.ConsumerMemberState; +import org.apache.kafka.coordinator.group.CompactionReplayTestContext.StreamsMemberState; +import org.apache.kafka.coordinator.group.Group.GroupType; +import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics; +import org.apache.kafka.coordinator.group.streams.MockTaskAssignor; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment; +import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment; +import static org.apache.kafka.coordinator.group.CompactionReplayTestContext.BAR_TOPIC_NAME; +import static org.apache.kafka.coordinator.group.CompactionReplayTestContext.FOO_TOPIC_NAME; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Compaction replay tests for the group coordinator. Tests capture written records, + * compact the resulting log, and replay records through a new group coordinator + * shard to verify loading. + * + * Two compaction cases are tested: + * - Prefix compaction (standard): A prefix of the log is compacted, so during loading, + * the group coordinator reads a compacted section followed by an uncompacted section. + * - Concurrent compaction: When compaction occurs concurrent with a load, the group + * coordinator can read a compacted section in between uncompacted sections. + * More precisely, something like the following can happen - + * 1. Group coordinator loads segment A (uncompacted) + * 2. Sections A and B are compacted + * 3. Group coordinator loads sections B (compacted) then C (active, uncompacted) + * This scenario has been seen in production and caused KAFKA-19862. + * + * The compaction model in this test class aligns compaction to batch boundaries to match + * realistic compaction performance. Consider a scenario like the following: + * 1. ConsumerGroupMemberMetadataKey <- memberEpoch=0 + * ... + * 2. ConsumerGroupCurrentMemberAssignmentKey, compacted + * --- batch boundary --- + * 3. ConsumerGroupCurrentMemberAssignmentKey = tombstone, compacted + * 4. ConsumerGroupTargetAssignmentMemberKey = tombstone + * 5. ConsumerGroupMemberMetadataKey = tombstone + * + * If we were to compact records 2,3 across a batch boundary, then record 5 will fail on load + * because it will see memberEpoch=0 (from record 1) but expect LEAVE_GROUP_MEMBER_EPOCH. Put + * another way, the batch boundaries mean that either all or no records for a given group + * operation are compacted. + */ +public class GroupCoordinatorShardCompactionReplayTest { + + private static final GroupCoordinatorConfig REPLAY_CONFIG = GroupCoordinatorConfig.fromProps(Map.of()); + private static final GroupCoordinatorMetrics REPLAY_METRICS = new GroupCoordinatorMetrics(); + + private Uuid fooTopicId; + private Uuid barTopicId; + private CoordinatorMetadataImage metadataImage; + + @BeforeEach + public void setUp() { + fooTopicId = Uuid.randomUuid(); + barTopicId = Uuid.randomUuid(); + metadataImage = new MetadataImageBuilder() + .addTopic(fooTopicId, FOO_TOPIC_NAME, 6) + .addTopic(barTopicId, BAR_TOPIC_NAME, 3) + .addRacks() + .buildCoordinatorMetadataImage(); + } + + private CompactionReplayTestContext newContext() { + MockPartitionAssignor consumerAssignor = new MockPartitionAssignor("range"); + MockTaskAssignor streamsAssignor = new MockTaskAssignor("sticky"); + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.BIDIRECTIONAL.toString()) + .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(consumerAssignor)) + .withStreamsGroupTaskAssignors(List.of(streamsAssignor)) + .withMetadataImage(metadataImage) + .build(); + return new CompactionReplayTestContext(context, consumerAssignor, streamsAssignor, metadataImage); + } + + /** + * Classic -> consumer group upgrade with offset commits. + * + * Scenario: + * Classic group created + * Classic offset commit + * Classic group rebalance + * Member joins with consumer protocol + * Upgrades to consumer group + * Consumer group rebalance + */ + @Test + public void testClassicGroupUpgradeToConsumerGroup() throws Exception { + CompactionReplayTestContext context = newContext(); + String groupId = "consumer-lifecycle-group"; + + // A classic group is created when its first member joins and syncs + JoinGroupResponseData joinResponseA = context.joinFirstClassicMember(groupId); + String classicMemberA = joinResponseA.memberId(); + context.syncClassicMember(groupId, classicMemberA, joinResponseA.generationId(), Map.of( + classicMemberA, List.of( + new TopicPartition(FOO_TOPIC_NAME, 0), + new TopicPartition(FOO_TOPIC_NAME, 1), + new TopicPartition(FOO_TOPIC_NAME, 2), + new TopicPartition(FOO_TOPIC_NAME, 3), + new TopicPartition(FOO_TOPIC_NAME, 4), + new TopicPartition(FOO_TOPIC_NAME, 5), + new TopicPartition(BAR_TOPIC_NAME, 0), + new TopicPartition(BAR_TOPIC_NAME, 1), + new TopicPartition(BAR_TOPIC_NAME, 2)) + )); + + // Offset commit + context.commitOffset(groupId, FOO_TOPIC_NAME, 0, 10L); + context.commitOffset(groupId, BAR_TOPIC_NAME, 0, 20L); + + // Member B joins with classic protocol, triggering rebalance + String classicMemberB = context.joinClassicMember(groupId); + + // Member A rejoins + JoinGroupResponseData rejoinResponseA = context.rejoinClassicMember(groupId, classicMemberA); + context.syncClassicMember(groupId, classicMemberA, rejoinResponseA.generationId(), Map.of( + classicMemberA, List.of( + new TopicPartition(FOO_TOPIC_NAME, 0), + new TopicPartition(FOO_TOPIC_NAME, 1), + new TopicPartition(FOO_TOPIC_NAME, 2), + new TopicPartition(BAR_TOPIC_NAME, 0)), + classicMemberB, List.of( + new TopicPartition(FOO_TOPIC_NAME, 3), + new TopicPartition(FOO_TOPIC_NAME, 4), + new TopicPartition(FOO_TOPIC_NAME, 5), + new TopicPartition(BAR_TOPIC_NAME, 1), + new TopicPartition(BAR_TOPIC_NAME, 2)) + )); + context.syncClassicMember(groupId, classicMemberB, rejoinResponseA.generationId(), Map.of()); + + // Member C joins with consumer protocol, triggering an online classic -> consumer group upgrade. + String memberC = Uuid.randomUuid().toString(); + context.prepareConsumerAssignment(Map.of( + classicMemberA, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1), mkTopicAssignment(barTopicId, 0)), + classicMemberB, mkAssignment(mkTopicAssignment(fooTopicId, 2, 3), mkTopicAssignment(barTopicId, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + Map<String, ConsumerMemberState> members = new LinkedHashMap<>(); + context.joinConsumerMember(groupId, memberC, members); + + // Members A and B move onto the consumer protocol one at a time. + String memberA = Uuid.randomUuid().toString(); + context.prepareConsumerAssignment(Map.of( + classicMemberB, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3), mkTopicAssignment(barTopicId, 0, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + context.leaveClassicMember(groupId, classicMemberA); + context.prepareConsumerAssignment(Map.of( + classicMemberB, mkAssignment(mkTopicAssignment(fooTopicId, 2, 3), mkTopicAssignment(barTopicId, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)), + memberA, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1), mkTopicAssignment(barTopicId, 0)))); + context.waitForAssignmentInterval(); + context.joinConsumerMember(groupId, memberA, members); + context.completeConsumerGroupRebalance(groupId, members); + + String memberB = Uuid.randomUuid().toString(); + context.prepareConsumerAssignment(Map.of( + memberA, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3), mkTopicAssignment(barTopicId, 0, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + context.leaveClassicMember(groupId, classicMemberB); + context.completeConsumerGroupRebalance(groupId, members); + context.prepareConsumerAssignment(Map.of( + memberA, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1), mkTopicAssignment(barTopicId, 0)), + memberB, mkAssignment(mkTopicAssignment(fooTopicId, 2, 3), mkTopicAssignment(barTopicId, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + context.waitForAssignmentInterval(); + context.joinConsumerMember(groupId, memberB, members); + context.completeConsumerGroupRebalance(groupId, members); + + // Member D joins with consumer protocol, triggering a consumer group rebalance. + String memberD = Uuid.randomUuid().toString(); + context.prepareConsumerAssignment(Map.of( + memberA, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5)), + memberB, mkAssignment(mkTopicAssignment(fooTopicId, 0), mkTopicAssignment(barTopicId, 0)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 1), mkTopicAssignment(barTopicId, 1)), + memberD, mkAssignment(mkTopicAssignment(fooTopicId, 2, 3), mkTopicAssignment(barTopicId, 2)))); + context.waitForAssignmentInterval(); + context.joinConsumerMember(groupId, memberD, members); + context.completeConsumerGroupRebalance(groupId, members); + + // Group commits one more offset + context.commitOffset(groupId, FOO_TOPIC_NAME, 1, 30L); + + // Verify the partitions can be reloaded cleanly from log. + assertCompactedVariantsLoadCleanly(context); + } + + /** + * Classic -> streams upgrade with offset commits. Related bugs: KAFKA-20254 + * + * Scenario: + * Classic group created + * Classic offset commit + * Classic group rebalance + * Group upgrades offline to streams protocol + * Members join/leave and group rebalances accordingly + */ + @Test + public void testClassicGroupMigratedToStreamsGroup() throws Exception { + CompactionReplayTestContext context = newContext(); + String groupId = "streams-lifecycle-group"; + + // A classic group is created when its first member joins and syncs + JoinGroupResponseData joinResponseA = context.joinFirstClassicMember(groupId); + String classicMemberA = joinResponseA.memberId(); + context.syncClassicMember(groupId, classicMemberA, joinResponseA.generationId(), Map.of( + classicMemberA, List.of( + new TopicPartition(FOO_TOPIC_NAME, 0), + new TopicPartition(FOO_TOPIC_NAME, 1), + new TopicPartition(FOO_TOPIC_NAME, 2), + new TopicPartition(FOO_TOPIC_NAME, 3), + new TopicPartition(FOO_TOPIC_NAME, 4), + new TopicPartition(FOO_TOPIC_NAME, 5)) + )); + + // Offset commit + context.commitOffset(groupId, FOO_TOPIC_NAME, 0, 10L); + context.commitOffset(groupId, FOO_TOPIC_NAME, 1, 20L); + + // Member B joins with classic protocol, triggering rebalance + String classicMemberB = context.joinClassicMember(groupId); + + // Member A rejoins + JoinGroupResponseData rejoinResponseA = context.rejoinClassicMember(groupId, classicMemberA); + context.syncClassicMember(groupId, classicMemberA, rejoinResponseA.generationId(), Map.of( + classicMemberA, List.of( + new TopicPartition(FOO_TOPIC_NAME, 0), + new TopicPartition(FOO_TOPIC_NAME, 1), + new TopicPartition(FOO_TOPIC_NAME, 2)), + classicMemberB, List.of( + new TopicPartition(FOO_TOPIC_NAME, 3), + new TopicPartition(FOO_TOPIC_NAME, 4), + new TopicPartition(FOO_TOPIC_NAME, 5)) + )); + context.syncClassicMember(groupId, classicMemberB, rejoinResponseA.generationId(), Map.of()); + + // Group is shut down for offline upgrade to streams + context.leaveClassicMember(groupId, classicMemberA); + context.leaveClassicMember(groupId, classicMemberB); + + // Group restarts with streams protocol. The leftover classic group is tombstoned. + String streamsMemberA = Uuid.randomUuid().toString(); + context.prepareStreamsAssignment(Map.of(streamsMemberA, context.tasks(0, 1, 2, 3, 4, 5))); + Map<String, StreamsMemberState> members = new LinkedHashMap<>(); + context.joinStreamsMember(groupId, streamsMemberA, "process-a", members); + context.completeStreamsGroupRebalance(groupId, members); + + // Member B joins and group rebalances. + String streamsMemberB = Uuid.randomUuid().toString(); + context.prepareStreamsAssignment(Map.of( + streamsMemberA, context.tasks(0, 1, 2), + streamsMemberB, context.tasks(3, 4, 5))); + context.waitForAssignmentInterval(); + context.joinStreamsMember(groupId, streamsMemberB, "process-b", members); + context.completeStreamsGroupRebalance(groupId, members); + + // Member C joins and the group rebalances. + String streamsMemberC = Uuid.randomUuid().toString(); + context.prepareStreamsAssignment(Map.of( + streamsMemberA, context.tasks(0, 1), + streamsMemberB, context.tasks(2, 3), + streamsMemberC, context.tasks(4, 5))); + context.waitForAssignmentInterval(); + context.joinStreamsMember(groupId, streamsMemberC, "process-c", members); + context.completeStreamsGroupRebalance(groupId, members); + + // Member A leaves and the group rebalances. + context.prepareStreamsAssignment(Map.of( + streamsMemberB, context.tasks(0, 1, 2), + streamsMemberC, context.tasks(3, 4, 5))); + context.waitForAssignmentInterval(); + context.leaveStreamsMember(groupId, streamsMemberA, members); + context.completeStreamsGroupRebalance(groupId, members); + + // Member B leaves and group rebalances (all tasks now owned by member C). + context.prepareStreamsAssignment(Map.of(streamsMemberC, context.tasks(0, 1, 2, 3, 4, 5))); + context.waitForAssignmentInterval(); + context.leaveStreamsMember(groupId, streamsMemberB, members); + context.completeStreamsGroupRebalance(groupId, members); + + context.commitOffset(groupId, FOO_TOPIC_NAME, 2, 30L); + + // Verify partitions can be reloaded cleanly from log. + assertCompactedVariantsLoadCleanly(context); + } + + /** + * Consumer -> classic downgrade by leave. + * + * Scenario: + * Classic group created and rebalanced + * Member joins with consumer protocol, upgrading the group to a consumer group + * The last consumer-protocol member leaves, downgrading the group back to classic + * Classic group commits an offset + */ + @Test + @Disabled // Enable when KAFKA-21102 is fixed + public void testConsumerGroupDowngradeByLeave() throws Exception { + CompactionReplayTestContext context = newContext(); + String groupId = "consumer-downgrade-by-leave-group"; + + // A classic group is created when its first member joins and syncs + JoinGroupResponseData joinResponseA = context.joinFirstClassicMember(groupId); + String classicMemberA = joinResponseA.memberId(); + context.syncClassicMember(groupId, classicMemberA, joinResponseA.generationId(), Map.of( + classicMemberA, List.of( + new TopicPartition(FOO_TOPIC_NAME, 0), + new TopicPartition(FOO_TOPIC_NAME, 1), + new TopicPartition(FOO_TOPIC_NAME, 2), + new TopicPartition(FOO_TOPIC_NAME, 3), + new TopicPartition(FOO_TOPIC_NAME, 4), + new TopicPartition(FOO_TOPIC_NAME, 5), + new TopicPartition(BAR_TOPIC_NAME, 0), + new TopicPartition(BAR_TOPIC_NAME, 1), + new TopicPartition(BAR_TOPIC_NAME, 2)) + )); + + // Member B joins with classic protocol, triggering rebalance + String classicMemberB = context.joinClassicMember(groupId); + + // Member A rejoins + JoinGroupResponseData rejoinResponseA = context.rejoinClassicMember(groupId, classicMemberA); + context.syncClassicMember(groupId, classicMemberA, rejoinResponseA.generationId(), Map.of( + classicMemberA, List.of( + new TopicPartition(FOO_TOPIC_NAME, 0), + new TopicPartition(FOO_TOPIC_NAME, 1), + new TopicPartition(FOO_TOPIC_NAME, 2), + new TopicPartition(BAR_TOPIC_NAME, 0)), + classicMemberB, List.of( + new TopicPartition(FOO_TOPIC_NAME, 3), + new TopicPartition(FOO_TOPIC_NAME, 4), + new TopicPartition(FOO_TOPIC_NAME, 5), + new TopicPartition(BAR_TOPIC_NAME, 1), + new TopicPartition(BAR_TOPIC_NAME, 2)) + )); + context.syncClassicMember(groupId, classicMemberB, rejoinResponseA.generationId(), Map.of()); + + // Member C joins with the consumer protocol, upgrading the group online to a consumer group. + // Members A and B stay on the classic protocol. + String memberC = Uuid.randomUuid().toString(); + context.prepareConsumerAssignment(Map.of( + classicMemberA, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2), mkTopicAssignment(barTopicId, 0)), + classicMemberB, mkAssignment(mkTopicAssignment(fooTopicId, 3, 4, 5), mkTopicAssignment(barTopicId, 1, 2)))); + Map<String, ConsumerMemberState> members = new LinkedHashMap<>(); + context.joinConsumerMember(groupId, memberC, members); + assertEquals(GroupType.CONSUMER, context.groupType(groupId)); + + // Member C, the last consumer-protocol member, leaves; the group downgrades back to classic + // with members A and B. + context.leaveConsumerMember(groupId, memberC, members); + assertEquals(GroupType.CLASSIC, context.groupType(groupId)); + + // The classic group keeps working and commits an offset. + context.commitOffset(groupId, FOO_TOPIC_NAME, 0, 40L); + + // Verify partitions can be reloaded cleanly from log. + assertCompactedVariantsLoadCleanly(context); + } + + /** + * Consumer -> classic downgrade by static member replacement. + * + * Scenario: + * Classic group created + * Static member joins with consumer protocol, upgrading the group to a consumer group + * A classic member replaces the static consumer member, downgrading the group back to classic + * Classic group commits an offset + */ + @Test + @Disabled // Enable when KAFKA-21102 is fixed Review Comment: Heard thanks, I'll create the jira and update it. -- 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]
