squah-confluent commented on code in PR #23220: URL: https://github.com/apache/kafka/pull/23220#discussion_r3870095042
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerCompactionReplayTest.java: ########## @@ -0,0 +1,786 @@ +/* + * 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.clients.consumer.ConsumerPartitionAssignor; +import org.apache.kafka.clients.consumer.internals.ConsumerProtocol; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData; +import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData; +import org.apache.kafka.common.message.JoinGroupRequestData; +import org.apache.kafka.common.message.JoinGroupResponseData; +import org.apache.kafka.common.message.LeaveGroupRequestData; +import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData; +import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.Subtopology; +import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.Topology; +import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData; +import org.apache.kafka.common.message.SyncGroupRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.record.internal.RecordBatch; +import org.apache.kafka.common.utils.Utils; +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.api.assignor.GroupAssignment; +import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics; +import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl; +import org.apache.kafka.coordinator.group.streams.MockTaskAssignor; +import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil; +import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.TaskRole; +import org.apache.kafka.coordinator.group.streams.TasksTuple; + +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.OptionalInt; +import java.util.OptionalLong; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.kafka.common.requests.JoinGroupRequest.UNKNOWN_MEMBER_ID; +import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH; +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.StreamsGroupTestUtil.staticHeartbeat; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticJoinHeartbeat; + +/** + * Compaction replay tests for the group coordinator. + * + * Tests check partition loading after compaction for non-trivial scenarios involving offset + * commits and member joins/rebalances for an online classic -> consumer upgrade and an + * offline classic -> streams upgrade. Both tests capture written records, compact the + * resulting log, and replay records through a new group coordinator shard to verify loading. + * Multiple contiguous log segments are tested, including prefix and mid-log windows. + * + */ +public class GroupMetadataManagerCompactionReplayTest { + + private static final String FOO_TOPIC_NAME = "foo"; + private static final String BAR_TOPIC_NAME = "bar"; + private static final String SUBTOPOLOGY_ID = "subtopology-1"; + + private static final int MAX_RECONCILIATION_ROUNDS = 10; + private static final long ASSIGNMENT_INTERVAL_ADVANCE_MS = + GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_DEFAULT + 1; + private static final int LONG_TIMEOUT_MS = 60000; + + 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 MockPartitionAssignor consumerAssignor; + private MockTaskAssignor streamsAssignor; + private GroupMetadataManagerTestContext context; + + @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(); + consumerAssignor = new MockPartitionAssignor("range"); + streamsAssignor = new MockTaskAssignor("sticky"); + context = new GroupMetadataManagerTestContext.Builder() + .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString()) + .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(consumerAssignor)) + .withStreamsGroupTaskAssignors(List.of(streamsAssignor)) + .withMetadataImage(metadataImage) + .build(); + } + + /** + * Classic -> consumer group upgrade with offset commits. Related bugs: KAFKA-19862 + * + * 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 testClassicGroupUpgradeToConsumerGroupWithOffsetCommit() throws Exception { + String groupId = "consumer-lifecycle-group"; + CapturedLog capturedLog = new CapturedLog(); + + // A classic group is created when its first member joins and syncs + JoinGroupResponseData joinResponseA = createClassicGroupWithFirstMember(groupId, capturedLog); + String classicMemberA = joinResponseA.memberId(); + 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)) + ), capturedLog); + + // Offset commit + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 0, 10L)); + capturedLog.append(offsetCommitRecord(groupId, BAR_TOPIC_NAME, 0, 20L)); + + // Member B joins with classic protocol, triggering rebalance + String classicMemberB = context.sendClassicGroupJoin( + classicJoinRequest(groupId, UNKNOWN_MEMBER_ID), true).joinFuture.get().memberId(); + context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberB), true); + + // Member A rejoins + var rejoinA = context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberA), true); + capturedLog.append(rejoinA.records); + JoinGroupResponseData rejoinResponseA = rejoinA.joinFuture.get(); + 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)) + ), capturedLog); + syncClassicMember(groupId, classicMemberB, rejoinResponseA.generationId(), Map.of(), capturedLog); + + // Member C joins with consumer protocol, triggering an online classic -> consumer group upgrade. + String memberC = Uuid.randomUuid().toString(); + 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<>(); + joinConsumerMember(groupId, memberC, members, capturedLog); + + // Members A and B move onto the consumer protocol one at a time. + String memberA = Uuid.randomUuid().toString(); + prepareConsumerAssignment(Map.of( + classicMemberB, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3), mkTopicAssignment(barTopicId, 0, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + leaveClassicMember(groupId, classicMemberA, capturedLog); + 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)))); + letAssignmentIntervalElapse(capturedLog); + joinConsumerMember(groupId, memberA, members, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + + String memberB = Uuid.randomUuid().toString(); + prepareConsumerAssignment(Map.of( + memberA, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3), mkTopicAssignment(barTopicId, 0, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + leaveClassicMember(groupId, classicMemberB, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + 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)))); + letAssignmentIntervalElapse(capturedLog); + joinConsumerMember(groupId, memberB, members, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + + // Member D joins with consumer protocol, triggering a consumer group rebalance. + String memberD = Uuid.randomUuid().toString(); + 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)))); + letAssignmentIntervalElapse(capturedLog); + joinConsumerMember(groupId, memberD, members, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + + // Group commits one more offset + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 1, 30L)); + + // Verify the partitions can be reloaded cleanly from log. + assertCompactedVariantsLoadCleanly(capturedLog); + } + + /** + * Classic -> streams upgrade with offset commits. Related bugs: KAFKA-19862, 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 testClassicGroupMigratedToStreamsGroupLoadsCleanlyUnderCompaction() throws Exception { + String groupId = "streams-lifecycle-group"; + CapturedLog capturedLog = new CapturedLog(); + + // A classic group is created when its first member joins and syncs + JoinGroupResponseData joinResponseA = createClassicGroupWithFirstMember(groupId, capturedLog); + String classicMemberA = joinResponseA.memberId(); + 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)) + ), capturedLog); + + // Offset commit + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 0, 10L)); + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 1, 20L)); + + // Member B joins with classic protocol, triggering rebalance + String classicMemberB = context.sendClassicGroupJoin( + classicJoinRequest(groupId, UNKNOWN_MEMBER_ID), true).joinFuture.get().memberId(); + context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberB), true); + + // Member A rejoins + var rejoinA = context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberA), true); + capturedLog.append(rejoinA.records); + JoinGroupResponseData rejoinResponseA = rejoinA.joinFuture.get(); + 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)) + ), capturedLog); + syncClassicMember(groupId, classicMemberB, rejoinResponseA.generationId(), Map.of(), capturedLog); + + // Group is shut down for offline upgrade to streams + leaveClassicMember(groupId, classicMemberA, capturedLog); + leaveClassicMember(groupId, classicMemberB, capturedLog); + + // Group restarts with streams protocol. The leftover classic group is tombstoned. + String streamsMemberA = Uuid.randomUuid().toString(); + streamsAssignor.prepareGroupAssignment(Map.of(streamsMemberA, tasks(0, 1, 2, 3, 4, 5))); + Map<String, StreamsMemberState> members = new LinkedHashMap<>(); + joinStreamsMember(groupId, streamsMemberA, "process-a", members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member B joins and group rebalances. + String streamsMemberB = Uuid.randomUuid().toString(); + streamsAssignor.prepareGroupAssignment(Map.of( + streamsMemberA, tasks(0, 1, 2), + streamsMemberB, tasks(3, 4, 5))); + letAssignmentIntervalElapse(capturedLog); + joinStreamsMember(groupId, streamsMemberB, "process-b", members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member C joins and the group rebalances. + String streamsMemberC = Uuid.randomUuid().toString(); + streamsAssignor.prepareGroupAssignment(Map.of( + streamsMemberA, tasks(0, 1), + streamsMemberB, tasks(2, 3), + streamsMemberC, tasks(4, 5))); + letAssignmentIntervalElapse(capturedLog); + joinStreamsMember(groupId, streamsMemberC, "process-c", members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member A leaves and the group rebalances. + streamsAssignor.prepareGroupAssignment(Map.of( + streamsMemberB, tasks(0, 1, 2), + streamsMemberC, tasks(3, 4, 5))); + letAssignmentIntervalElapse(capturedLog); + leaveStreamsMember(groupId, streamsMemberA, members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member B leaves and group rebalances (all tasks now owned by member C). + streamsAssignor.prepareGroupAssignment(Map.of(streamsMemberC, tasks(0, 1, 2, 3, 4, 5))); + letAssignmentIntervalElapse(capturedLog); + leaveStreamsMember(groupId, streamsMemberB, members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 2, 30L)); + + // Verify partitions can be reloaded cleanly from log. + assertCompactedVariantsLoadCleanly(capturedLog); + } + + // ------------------------------------------------------------------------------------------ + // Scenario helpers. + // ------------------------------------------------------------------------------------------ + + /** + * A classic join request using the consumer embedded protocol. + */ + private JoinGroupRequestData classicJoinRequest(String groupId, String memberId) { + return new GroupMetadataManagerTestContext.JoinGroupRequestBuilder() + .withGroupId(groupId) + .withMemberId(memberId) + .withProtocolType("consumer") + .withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol( + List.of(FOO_TOPIC_NAME, BAR_TOPIC_NAME), List.of())) + .withRebalanceTimeoutMs(LONG_TIMEOUT_MS) + .withSessionTimeoutMs(LONG_TIMEOUT_MS) + .build(); + } + + /** + * Creates a classic group when the first member joins. + */ + private JoinGroupResponseData createClassicGroupWithFirstMember( + String groupId, + CapturedLog capturedLog + ) throws Exception { + var firstJoin = context.sendClassicGroupJoin(classicJoinRequest(groupId, UNKNOWN_MEMBER_ID), true); + capturedLog.append(firstJoin.records); + firstJoin.appendFuture.complete(null); + String memberId = firstJoin.joinFuture.get().memberId(); + + var secondJoin = context.sendClassicGroupJoin(classicJoinRequest(groupId, memberId), true); + capturedLog.append(secondJoin.records); + // The first generation only forms once the initial rebalance delay has elapsed. + sleepCapturing(context.classicGroupInitialRebalanceDelayMs, capturedLog); + return secondJoin.joinFuture.get(); + } + + /** + * Advances the clock, capturing whatever the timeouts that fired wrote. + */ + private void sleepCapturing(long durationMs, CapturedLog capturedLog) { + context.sleep(durationMs).forEach(timeout -> capturedLog.append(timeout.result().records())); + } + + /** + * Advances the clock past the assignment interval so that the next heartbeat is allowed to run the + * assignor and move partitions or tasks between members. + */ + private void letAssignmentIntervalElapse(CapturedLog capturedLog) { Review Comment: It's unusual to use `let` as a verb in a method name. `sleepForAssignmentInterval` or `waitForAssignmentInterval`? Also technically this is the consumer group assignment interval. It just happens to be the same for streams. We could take the max interval over all protocols instead. ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerCompactionReplayTest.java: ########## @@ -0,0 +1,786 @@ +/* + * 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.clients.consumer.ConsumerPartitionAssignor; +import org.apache.kafka.clients.consumer.internals.ConsumerProtocol; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData; +import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData; +import org.apache.kafka.common.message.JoinGroupRequestData; +import org.apache.kafka.common.message.JoinGroupResponseData; +import org.apache.kafka.common.message.LeaveGroupRequestData; +import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData; +import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.Subtopology; +import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.Topology; +import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData; +import org.apache.kafka.common.message.SyncGroupRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.record.internal.RecordBatch; +import org.apache.kafka.common.utils.Utils; +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.api.assignor.GroupAssignment; +import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics; +import org.apache.kafka.coordinator.group.modern.MemberAssignmentImpl; +import org.apache.kafka.coordinator.group.streams.MockTaskAssignor; +import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil; +import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.TaskRole; +import org.apache.kafka.coordinator.group.streams.TasksTuple; + +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.OptionalInt; +import java.util.OptionalLong; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.kafka.common.requests.JoinGroupRequest.UNKNOWN_MEMBER_ID; +import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH; +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.StreamsGroupTestUtil.staticHeartbeat; +import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.staticJoinHeartbeat; + +/** + * Compaction replay tests for the group coordinator. + * + * Tests check partition loading after compaction for non-trivial scenarios involving offset + * commits and member joins/rebalances for an online classic -> consumer upgrade and an + * offline classic -> streams upgrade. Both tests capture written records, compact the + * resulting log, and replay records through a new group coordinator shard to verify loading. + * Multiple contiguous log segments are tested, including prefix and mid-log windows. + * + */ +public class GroupMetadataManagerCompactionReplayTest { + + private static final String FOO_TOPIC_NAME = "foo"; + private static final String BAR_TOPIC_NAME = "bar"; + private static final String SUBTOPOLOGY_ID = "subtopology-1"; + + private static final int MAX_RECONCILIATION_ROUNDS = 10; + private static final long ASSIGNMENT_INTERVAL_ADVANCE_MS = + GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_DEFAULT + 1; + private static final int LONG_TIMEOUT_MS = 60000; + + 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 MockPartitionAssignor consumerAssignor; + private MockTaskAssignor streamsAssignor; + private GroupMetadataManagerTestContext context; + + @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(); + consumerAssignor = new MockPartitionAssignor("range"); + streamsAssignor = new MockTaskAssignor("sticky"); + context = new GroupMetadataManagerTestContext.Builder() + .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString()) + .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(consumerAssignor)) + .withStreamsGroupTaskAssignors(List.of(streamsAssignor)) + .withMetadataImage(metadataImage) + .build(); + } + + /** + * Classic -> consumer group upgrade with offset commits. Related bugs: KAFKA-19862 + * + * 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 testClassicGroupUpgradeToConsumerGroupWithOffsetCommit() throws Exception { + String groupId = "consumer-lifecycle-group"; + CapturedLog capturedLog = new CapturedLog(); + + // A classic group is created when its first member joins and syncs + JoinGroupResponseData joinResponseA = createClassicGroupWithFirstMember(groupId, capturedLog); + String classicMemberA = joinResponseA.memberId(); + 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)) + ), capturedLog); + + // Offset commit + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 0, 10L)); + capturedLog.append(offsetCommitRecord(groupId, BAR_TOPIC_NAME, 0, 20L)); + + // Member B joins with classic protocol, triggering rebalance + String classicMemberB = context.sendClassicGroupJoin( + classicJoinRequest(groupId, UNKNOWN_MEMBER_ID), true).joinFuture.get().memberId(); + context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberB), true); + + // Member A rejoins + var rejoinA = context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberA), true); + capturedLog.append(rejoinA.records); + JoinGroupResponseData rejoinResponseA = rejoinA.joinFuture.get(); + 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)) + ), capturedLog); + syncClassicMember(groupId, classicMemberB, rejoinResponseA.generationId(), Map.of(), capturedLog); + + // Member C joins with consumer protocol, triggering an online classic -> consumer group upgrade. + String memberC = Uuid.randomUuid().toString(); + 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<>(); + joinConsumerMember(groupId, memberC, members, capturedLog); + + // Members A and B move onto the consumer protocol one at a time. + String memberA = Uuid.randomUuid().toString(); + prepareConsumerAssignment(Map.of( + classicMemberB, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3), mkTopicAssignment(barTopicId, 0, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + leaveClassicMember(groupId, classicMemberA, capturedLog); + 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)))); + letAssignmentIntervalElapse(capturedLog); + joinConsumerMember(groupId, memberA, members, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + + String memberB = Uuid.randomUuid().toString(); + prepareConsumerAssignment(Map.of( + memberA, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3), mkTopicAssignment(barTopicId, 0, 1)), + memberC, mkAssignment(mkTopicAssignment(fooTopicId, 4, 5), mkTopicAssignment(barTopicId, 2)))); + leaveClassicMember(groupId, classicMemberB, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + 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)))); + letAssignmentIntervalElapse(capturedLog); + joinConsumerMember(groupId, memberB, members, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + + // Member D joins with consumer protocol, triggering a consumer group rebalance. + String memberD = Uuid.randomUuid().toString(); + 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)))); + letAssignmentIntervalElapse(capturedLog); + joinConsumerMember(groupId, memberD, members, capturedLog); + completeConsumerGroupRebalance(groupId, members, capturedLog); + + // Group commits one more offset + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 1, 30L)); + + // Verify the partitions can be reloaded cleanly from log. + assertCompactedVariantsLoadCleanly(capturedLog); + } + + /** + * Classic -> streams upgrade with offset commits. Related bugs: KAFKA-19862, 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 testClassicGroupMigratedToStreamsGroupLoadsCleanlyUnderCompaction() throws Exception { + String groupId = "streams-lifecycle-group"; + CapturedLog capturedLog = new CapturedLog(); + + // A classic group is created when its first member joins and syncs + JoinGroupResponseData joinResponseA = createClassicGroupWithFirstMember(groupId, capturedLog); + String classicMemberA = joinResponseA.memberId(); + 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)) + ), capturedLog); + + // Offset commit + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 0, 10L)); + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 1, 20L)); + + // Member B joins with classic protocol, triggering rebalance + String classicMemberB = context.sendClassicGroupJoin( + classicJoinRequest(groupId, UNKNOWN_MEMBER_ID), true).joinFuture.get().memberId(); + context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberB), true); + + // Member A rejoins + var rejoinA = context.sendClassicGroupJoin(classicJoinRequest(groupId, classicMemberA), true); + capturedLog.append(rejoinA.records); + JoinGroupResponseData rejoinResponseA = rejoinA.joinFuture.get(); + 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)) + ), capturedLog); + syncClassicMember(groupId, classicMemberB, rejoinResponseA.generationId(), Map.of(), capturedLog); + + // Group is shut down for offline upgrade to streams + leaveClassicMember(groupId, classicMemberA, capturedLog); + leaveClassicMember(groupId, classicMemberB, capturedLog); + + // Group restarts with streams protocol. The leftover classic group is tombstoned. + String streamsMemberA = Uuid.randomUuid().toString(); + streamsAssignor.prepareGroupAssignment(Map.of(streamsMemberA, tasks(0, 1, 2, 3, 4, 5))); + Map<String, StreamsMemberState> members = new LinkedHashMap<>(); + joinStreamsMember(groupId, streamsMemberA, "process-a", members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member B joins and group rebalances. + String streamsMemberB = Uuid.randomUuid().toString(); + streamsAssignor.prepareGroupAssignment(Map.of( + streamsMemberA, tasks(0, 1, 2), + streamsMemberB, tasks(3, 4, 5))); + letAssignmentIntervalElapse(capturedLog); + joinStreamsMember(groupId, streamsMemberB, "process-b", members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member C joins and the group rebalances. + String streamsMemberC = Uuid.randomUuid().toString(); + streamsAssignor.prepareGroupAssignment(Map.of( + streamsMemberA, tasks(0, 1), + streamsMemberB, tasks(2, 3), + streamsMemberC, tasks(4, 5))); + letAssignmentIntervalElapse(capturedLog); + joinStreamsMember(groupId, streamsMemberC, "process-c", members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member A leaves and the group rebalances. + streamsAssignor.prepareGroupAssignment(Map.of( + streamsMemberB, tasks(0, 1, 2), + streamsMemberC, tasks(3, 4, 5))); + letAssignmentIntervalElapse(capturedLog); + leaveStreamsMember(groupId, streamsMemberA, members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + // Member B leaves and group rebalances (all tasks now owned by member C). + streamsAssignor.prepareGroupAssignment(Map.of(streamsMemberC, tasks(0, 1, 2, 3, 4, 5))); + letAssignmentIntervalElapse(capturedLog); + leaveStreamsMember(groupId, streamsMemberB, members, capturedLog); + completeStreamsGroupRebalance(groupId, members, capturedLog); + + capturedLog.append(offsetCommitRecord(groupId, FOO_TOPIC_NAME, 2, 30L)); + + // Verify partitions can be reloaded cleanly from log. + assertCompactedVariantsLoadCleanly(capturedLog); + } + + // ------------------------------------------------------------------------------------------ + // Scenario helpers. + // ------------------------------------------------------------------------------------------ Review Comment: We usually don't style comments like this. The grouping is very useful though. To keep things tidy, we could separate out these request helpers into a "...TestContext" class in a separate file, that also acts as the accumulator for records (inlining CapturedLog). -- 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]
