squah-confluent commented on code in PR #23220: URL: https://github.com/apache/kafka/pull/23220#discussion_r3981894654
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardCompactionReplayTest.java: ########## @@ -0,0 +1,596 @@ +/* + * 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.metrics.GroupCoordinatorMetrics; +import org.apache.kafka.coordinator.group.streams.MockTaskAssignor; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +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; + private CompactionReplayTestContext replay; + + @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(); + 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(); + replay = new CompactionReplayTestContext(context, consumerAssignor, streamsAssignor, metadataImage); + } + + /** + * Classic -> consumer group upgrade with offset commits. Related bugs: KAFKA-19862 Review Comment: Is this jira tagging accurate? I thought KAFKA-19862 didn't require an upgrade or downgrade. -- 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]
