content: /*
   * 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.modern.share;
  
  import org.apache.kafka.common.Uuid;
  import org.apache.kafka.common.errors.GroupIdNotFoundException;
  import org.apache.kafka.common.errors.GroupNotEmptyException;
  import org.apache.kafka.common.errors.UnknownMemberIdException;
  import org.apache.kafka.common.message.ShareGroupDescribeResponseData;
  import org.apache.kafka.common.protocol.ApiKeys;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.common.utils.MockTime;
  import org.apache.kafka.common.utils.annotation.ApiKeyVersionsSource;
  import org.apache.kafka.coordinator.group.Group;
  import org.apache.kafka.coordinator.group.MetadataImageBuilder;
  import org.apache.kafka.coordinator.group.modern.Assignment;
  import org.apache.kafka.coordinator.group.modern.MemberState;
  import org.apache.kafka.coordinator.group.modern.share.ShareGroup.ShareGroupState;
  import org.apache.kafka.timeline.SnapshotRegistry;
  
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.params.ParameterizedTest;
  
  import java.util.Arrays;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  
  import static org.apache.kafka.common.utils.Utils.mkEntry;
  import static org.apache.kafka.common.utils.Utils.mkMap;
  import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HETEROGENEOUS;
  import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
  import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  public class ShareGroupTest {
  
      @Test
      public void testType() {
          ShareGroup shareGroup = createShareGroup("foo");
          assertEquals(Group.GroupType.SHARE, shareGroup.type());
      }
  
      @Test
      public void testProtocolType() {
          ShareGroup shareGroup = createShareGroup("foo");
          assertEquals("share", shareGroup.protocolType());
      }
  
      @Test
      public void testGetOrCreateMember() {
          ShareGroup shareGroup = createShareGroup("foo");
          ShareGroupMember member;
  
          // Create a member.
          member = shareGroup.getOrMaybeCreateMember("member-id", true);
          assertEquals("member-id", member.memberId());
  
          // Add member to the group.
          shareGroup.updateMember(member);
  
          // Get that member back.
          member = shareGroup.getOrMaybeCreateMember("member-id", false);
          assertEquals("member-id", member.memberId());
  
          assertThrows(UnknownMemberIdException.class, () ->
              shareGroup.getOrMaybeCreateMember("does-not-exist", false));
      }
  
      @Test
      public void testUpdateMember() {
          ShareGroup shareGroup = createShareGroup("foo");
          ShareGroupMember member;
  
          member = shareGroup.getOrMaybeCreateMember("member", true);
  
          member = new ShareGroupMember.Builder(member)
              .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
              .build();
  
          shareGroup.updateMember(member);
  
          assertEquals(member, shareGroup.getOrMaybeCreateMember("member", false));
      }
  
      @Test
      public void testRemoveMember() {
          ShareGroup shareGroup = createShareGroup("foo");
  
          ShareGroupMember member = shareGroup.getOrMaybeCreateMember("member", true);
          shareGroup.updateMember(member);
          assertTrue(shareGroup.hasMember("member"));
  
          shareGroup.removeMember("member");
          assertFalse(shareGroup.hasMember("member"));
  
      }
  
      @Test
      public void testGroupState() {
          ShareGroup shareGroup = createShareGroup("foo");
          assertEquals(ShareGroup.ShareGroupState.EMPTY, shareGroup.state());
          assertEquals("Empty", shareGroup.stateAsString());
  
          ShareGroupMember member1 = new ShareGroupMember.Builder("member1")
              .setState(MemberState.STABLE)
              .setMemberEpoch(1)
              .setPreviousMemberEpoch(0)
              .build();
  
          shareGroup.updateMember(member1);
          shareGroup.setGroupEpoch(1);
  
          assertEquals(MemberState.STABLE, member1.state());
          assertEquals(ShareGroupState.STABLE, shareGroup.state());
          assertEquals("Stable", shareGroup.stateAsString());
      }
  
      @Test
      public void testGroupTypeFromString() {
          assertEquals(Group.GroupType.parse("share"), Group.GroupType.SHARE);
          // Test case insensitivity.
          assertEquals(Group.GroupType.parse("Share"), Group.GroupType.SHARE);
          assertEquals(Group.GroupType.parse("SHare"), Group.GroupType.SHARE);
      }
  
      @Test
      public void testUpdateSubscribedTopicNamesAndSubscriptionType() {
          ShareGroupMember member1 = new ShareGroupMember.Builder("member1")
              .setSubscribedTopicNames(List.of("foo"))
              .build();
          ShareGroupMember member2 = new ShareGroupMember.Builder("member2")
              .setSubscribedTopicNames(Arrays.asList("bar", "foo"))
              .build();
          ShareGroupMember member3 = new ShareGroupMember.Builder("member3")
              .setSubscribedTopicNames(Arrays.asList("bar", "foo"))
              .build();
  
          ShareGroup shareGroup = createShareGroup("group-foo");
  
          // It should be empty by default.
          assertEquals(
              Map.of(),
              shareGroup.subscribedTopicNames()
          );
  
          // It should be Homogeneous by default.
          assertEquals(
              HOMOGENEOUS,
              shareGroup.subscriptionType()
          );
  
          shareGroup.updateMember(member1);
  
          // It should be Homogeneous since there is just 1 member
          assertEquals(
              HOMOGENEOUS,
              shareGroup.subscriptionType()
          );
  
          shareGroup.updateMember(member2);
  
          assertEquals(
              HETEROGENEOUS,
              shareGroup.subscriptionType()
          );
  
          shareGroup.updateMember(member3);
  
          assertEquals(
              HETEROGENEOUS,
              shareGroup.subscriptionType()
          );
  
          shareGroup.removeMember(member1.memberId());
  
          assertEquals(
              HOMOGENEOUS,
              shareGroup.subscriptionType()
          );
  
          ShareGroupMember member4 = new ShareGroupMember.Builder("member2")
              .setSubscribedTopicNames(Arrays.asList("bar", "foo", "zar"))
              .build();
  
          shareGroup.updateMember(member4);
  
          assertEquals(
              HETEROGENEOUS,
              shareGroup.subscriptionType()
          );
      }
  
      @Test
      public void testUpdateInvertedAssignment() {
          SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
          ShareGroup shareGroup = new ShareGroup(snapshotRegistry, "test-group");
          Uuid topicId = Uuid.randomUuid();
          String memberId1 = "member1";
          String memberId2 = "member2";
  
          // Initial assignment for member1
          Assignment initialAssignment = new Assignment(Map.of(
              topicId,
              new HashSet<>(List.of(0))
          ));
          shareGroup.updateTargetAssignment(memberId1, initialAssignment);
  
          // Verify that partition 0 is assigned to member1.
          assertEquals(
              mkMap(
                  mkEntry(topicId, mkMap(mkEntry(0, memberId1)))
              ),
              shareGroup.invertedTargetAssignment()
          );
  
          // New assignment for member1
          Assignment newAssignment = new Assignment(Map.of(
              topicId,
              new HashSet<>(List.of(1))
          ));
          shareGroup.updateTargetAssignment(memberId1, newAssignment);
  
          // Verify that partition 0 is no longer assigned and partition 1 is assigned to member1
          assertEquals(
              mkMap(
                  mkEntry(topicId, mkMap(mkEntry(1, memberId1)))
              ),
              shareGroup.invertedTargetAssignment()
          );
  
          // New assignment for member2 to add partition 1
          Assignment newAssignment2 = new Assignment(Map.of(
              topicId,
              new HashSet<>(List.of(1))
          ));
          shareGroup.updateTargetAssignment(memberId2, newAssignment2);
  
          // Verify that partition 1 is assigned to member2
          assertEquals(
              mkMap(
                  mkEntry(topicId, mkMap(mkEntry(1, memberId2)))
              ),
              shareGroup.invertedTargetAssignment()
          );
  
          // New assignment for member1 to revoke partition 1 and assign partition 0
          Assignment newAssignment1 = new Assignment(Map.of(
              topicId,
              new HashSet<>(List.of(0))
          ));
          shareGroup.updateTargetAssignment(memberId1, newAssignment1);
  
          // Verify that partition 1 is still assigned to member2 and partition 0 is assigned to member1
          assertEquals(
              mkMap(
                  mkEntry(topicId, mkMap(
                      mkEntry(0, memberId1),
                      mkEntry(1, memberId2)
                  ))
              ),
              shareGroup.invertedTargetAssignment()
          );
  
          // Test remove target assignment for member1
          shareGroup.removeTargetAssignment(memberId1);
  
          // Verify that partition 0 is no longer assigned and partition 1 is still assigned to member2
          assertEquals(
              mkMap(
                  mkEntry(topicId, mkMap(mkEntry(1, memberId2)))
              ),
              shareGroup.invertedTargetAssignment()
          );
      }
  
      @Test
      public void testMetadataRefreshDeadline() {
          MockTime time = new MockTime();
          ShareGroup shareGroup = createShareGroup("group-foo");
  
          // Group epoch starts at 0.
          assertEquals(0, shareGroup.groupEpoch());
  
          // The refresh time deadline should be empty when the group is created or loaded.
          assertTrue(shareGroup.hasMetadataExpired(time.milliseconds()));
          assertEquals(0L, shareGroup.metadataRefreshDeadline().deadlineMs);
          assertEquals(0, shareGroup.metadataRefreshDeadline().epoch);
  
          // Set the refresh deadline. The metadata remains valid because the deadline
          // has not past and the group epoch is correct.
          shareGroup.setMetadataRefreshDeadline(time.milliseconds() + 1000, shareGroup.groupEpoch());
          assertFalse(shareGroup.hasMetadataExpired(time.milliseconds()));
          assertEquals(time.milliseconds() + 1000, shareGroup.metadataRefreshDeadline().deadlineMs);
          assertEquals(shareGroup.groupEpoch(), shareGroup.metadataRefreshDeadline().epoch);
  
          // Advance past the deadline. The metadata should have expired.
          time.sleep(1001L);
          assertTrue(shareGroup.hasMetadataExpired(time.milliseconds()));
  
          // Set the refresh time deadline with a higher group epoch. The metadata is considered
          // as expired because the group epoch attached to the deadline is higher than the
          // current group epoch.
          shareGroup.setMetadataRefreshDeadline(time.milliseconds() + 1000, shareGroup.groupEpoch() + 1);
          assertTrue(shareGroup.hasMetadataExpired(time.milliseconds()));
          assertEquals(time.milliseconds() + 1000, shareGroup.metadataRefreshDeadline().deadlineMs);
          assertEquals(shareGroup.groupEpoch() + 1, shareGroup.metadataRefreshDeadline().epoch);
  
          // Advance the group epoch.
          shareGroup.setGroupEpoch(shareGroup.groupEpoch() + 1);
  
          // Set the refresh deadline. The metadata remains valid because the deadline
          // has not past and the group epoch is correct.
          shareGroup.setMetadataRefreshDeadline(time.milliseconds() + 1000, shareGroup.groupEpoch());
          assertFalse(shareGroup.hasMetadataExpired(time.milliseconds()));
          assertEquals(time.milliseconds() + 1000, shareGroup.metadataRefreshDeadline().deadlineMs);
          assertEquals(shareGroup.groupEpoch(), shareGroup.metadataRefreshDeadline().epoch);
  
          // Request metadata refresh. The metadata expires immediately.
          shareGroup.requestMetadataRefresh();
          assertTrue(shareGroup.hasMetadataExpired(time.milliseconds()));
          assertEquals(0L, shareGroup.metadataRefreshDeadline().deadlineMs);
          assertEquals(0, shareGroup.metadataRefreshDeadline().epoch);
      }
  
      @ParameterizedTest
      @ApiKeyVersionsSource(apiKey = ApiKeys.OFFSET_COMMIT)
      public void testValidateOffsetCommit(short version) {
          ShareGroup shareGroup = createShareGroup("group-foo");
          assertThrows(GroupIdNotFoundException.class, () ->
              shareGroup.validateOffsetCommit(null, null, -1, false, version));
      }
  
      @Test
      public void testAsListedGroup() {
          SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
          ShareGroup shareGroup = new ShareGroup(snapshotRegistry, "group-foo");
          snapshotRegistry.idempotentCreateSnapshot(0);
          assertEquals(ShareGroupState.EMPTY, shareGroup.state(0));
          assertEquals("Empty", shareGroup.stateAsString(0));
          shareGroup.updateMember(new ShareGroupMember.Builder("member1")
              .setSubscribedTopicNames(List.of("foo"))
              .build());
          snapshotRegistry.idempotentCreateSnapshot(1);
          assertEquals(ShareGroupState.EMPTY, shareGroup.state(0));
          assertEquals("Empty", shareGroup.stateAsString(0));
          assertEquals(ShareGroupState.STABLE, shareGroup.state(1));
          assertEquals("Stable", shareGroup.stateAsString(1));
      }
  
      @Test
      public void testOffsetExpirationCondition() {
          ShareGroup shareGroup = createShareGroup("group-foo");
          assertThrows(UnsupportedOperationException.class, shareGroup::offsetExpirationCondition);
      }
  
      @Test
      public void testValidateOffsetFetch() {
          ShareGroup shareGroup = createShareGroup("group-foo");
          assertThrows(GroupIdNotFoundException.class, () ->
              shareGroup.validateOffsetFetch(null, -1, -1));
      }
  
      @Test
      public void testValidateOffsetDelete() {
          ShareGroup shareGroup = createShareGroup("group-foo");
          assertThrows(GroupIdNotFoundException.class, shareGroup::validateOffsetDelete);
      }
  
      @Test
      public void testValidateDeleteGroup() {
          ShareGroup shareGroup = createShareGroup("foo");
  
          assertEquals(ShareGroupState.EMPTY, shareGroup.state());
          assertDoesNotThrow(shareGroup::validateDeleteGroup);
  
          ShareGroupMember member1 = new ShareGroupMember.Builder("member1")
              .setMemberEpoch(1)
              .setPreviousMemberEpoch(0)
              .build();
          shareGroup.updateMember(member1);
  
          assertEquals(ShareGroupState.STABLE, shareGroup.state());
          assertThrows(GroupNotEmptyException.class, shareGroup::validateDeleteGroup);
  
          shareGroup.setGroupEpoch(1);
  
          assertEquals(ShareGroupState.STABLE, shareGroup.state());
          assertThrows(GroupNotEmptyException.class, shareGroup::validateDeleteGroup);
  
          shareGroup.setTargetAssignmentEpoch(1);
  
          assertEquals(ShareGroupState.STABLE, shareGroup.state());
          assertThrows(GroupNotEmptyException.class, shareGroup::validateDeleteGroup);
      }
  
      @Test
      public void testIsSubscribedToTopic() {
          ShareGroupMember member1 = new ShareGroupMember.Builder("member1")
              .setSubscribedTopicNames(List.of("foo"))
              .build();
          ShareGroupMember member2 = new ShareGroupMember.Builder("member2")
              .setSubscribedTopicNames(List.of("bar"))
              .build();
  
          ShareGroup shareGroup = createShareGroup("group-foo");
  
          shareGroup.updateMember(member1);
          shareGroup.updateMember(member2);
  
          assertTrue(shareGroup.isSubscribedToTopic("foo"));
          assertTrue(shareGroup.isSubscribedToTopic("bar"));
  
          shareGroup.removeMember("member1");
          assertFalse(shareGroup.isSubscribedToTopic("foo"));
  
          shareGroup.removeMember("member2");
          assertFalse(shareGroup.isSubscribedToTopic("bar"));
      }
  
      @Test
      public void testAsDescribedGroup() {
          SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
          ShareGroup shareGroup = new ShareGroup(snapshotRegistry, "group-id-1");
          snapshotRegistry.idempotentCreateSnapshot(0);
          assertEquals(ShareGroupState.EMPTY.toString(), shareGroup.stateAsString(0));
  
          shareGroup.updateMember(new ShareGroupMember.Builder("member1")
                  .setSubscribedTopicNames(List.of("foo"))
                  .build());
          shareGroup.updateMember(new ShareGroupMember.Builder("member2")
                  .build());
          snapshotRegistry.idempotentCreateSnapshot(1);
  
          ShareGroupDescribeResponseData.DescribedGroup expected = new ShareGroupDescribeResponseData.DescribedGroup()
              .setGroupId("group-id-1")
              .setGroupState(ShareGroupState.STABLE.toString())
              .setGroupEpoch(0)
              .setAssignmentEpoch(0)
              .setAssignorName("assignorName")
              .setMembers(Arrays.asList(
                  new ShareGroupDescribeResponseData.Member()
                      .setMemberId("member1")
                      .setSubscribedTopicNames(List.of("foo")),
                  new ShareGroupDescribeResponseData.Member().setMemberId("member2")
              ));
          ShareGroupDescribeResponseData.DescribedGroup actual = shareGroup.asDescribedGroup(1, "assignorName",
              new MetadataImageBuilder().build().topics());
  
          assertEquals(expected, actual);
      }
  
      @Test
      public void testIsInStatesCaseInsensitive() {
          SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
          ShareGroup shareGroup = new ShareGroup(snapshotRegistry, "group-foo");
          snapshotRegistry.idempotentCreateSnapshot(0);
          assertTrue(shareGroup.isInStates(Set.of("empty"), 0));
          assertFalse(shareGroup.isInStates(Set.of("Empty"), 0));
  
          shareGroup.updateMember(new ShareGroupMember.Builder("member1")
              .setSubscribedTopicNames(List.of("foo"))
              .build());
          snapshotRegistry.idempotentCreateSnapshot(1);
          assertTrue(shareGroup.isInStates(Set.of("empty"), 0));
          assertTrue(shareGroup.isInStates(Set.of("stable"), 1));
          assertFalse(shareGroup.isInStates(Set.of("empty"), 1));
      }
  
      private ShareGroup createShareGroup(String groupId) {
          SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
          return new ShareGroup(
              snapshotRegistry,
              groupId
          );
      }
  }
  