KAFKA-15598 & KAFKA-15461: Add integration tests for DescribeGroups A…

…PI, DeleteGroups API, OffsetDelete API and ListGroups API (apache#14537) This patch adds integration tests for four group coordinator APIs with new group coordinator and new protocol, new group coordinator and old protocol, and old group coordinator and old protocols for the following APIs: - DescribeGroups - DeleteGroups - OffsetDelete - ListGroups Reviewers: Ritika Reddy <[email protected]>, David Jacot <[email protected]>
tarunsankhla · Nov 2, 2023 · 5b37ec5 · 5b37ec5
1 parent b850b46
commit 5b37ec5
Show file tree

Hide file tree

Showing 6 changed files with 946 additions and 34 deletions.
diff --git a/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala b/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala
@@ -0,0 +1,135 @@
+/**
+ * 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 kafka.server
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.message.DescribeGroupsResponseData.DescribedGroup
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.coordinator.group.generic.GenericGroupState
+import org.junit.jupiter.api.Assertions.{assertEquals, fail}
+import org.junit.jupiter.api.{Tag, Timeout}
+import org.junit.jupiter.api.extension.ExtendWith
+
+@Timeout(120)
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(clusterType = Type.KRAFT, brokers = 1)
+@Tag("integration")
+class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
+  @ClusterTest(serverProperties = Array(
+    new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
+    new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+    new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
+    new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
+    new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+    new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+  ))
+  def testDeleteGroupsWithNewConsumerGroupProtocolAndNewGroupCoordinator(): Unit = {
+    testDeleteGroups(true)
+  }
+
+  @ClusterTest(serverProperties = Array(
+    new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
+    new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+    new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+    new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+  ))
+  def testDeleteGroupsWithOldConsumerGroupProtocolAndNewGroupCoordinator(): Unit = {
+    testDeleteGroups(false)
+  }
+
+  @ClusterTest(clusterType = Type.ALL, serverProperties = Array(
+    new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "false"),
+    new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "false"),
+    new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+    new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+  ))
+  def testDeleteGroupsWithOldConsumerGroupProtocolAndOldGroupCoordinator(): Unit = {
+    testDeleteGroups(false)
+  }
+
+  private def testDeleteGroups(useNewProtocol: Boolean): Unit = {
+    if (useNewProtocol && !isNewGroupCoordinatorEnabled) {
+      fail("Cannot use the new protocol with the old group coordinator.")
+    }
+
+    // Creates the __consumer_offsets topics because it won't be created automatically
+    // in this test because it does not use FindCoordinator API.
+    createOffsetsTopic()
+
+    // Create the topic.
+    createTopic(
+      topic = "foo",
+      numPartitions = 3
+    )
+
+    // Join the consumer group. Note that we don't heartbeat here so we must use
+    // a session long enough for the duration of the test.
+    // We test DeleteGroups on empty and non-empty groups. Here we create the non-empty group.
+    joinConsumerGroup(
+      groupId = "grp-non-empty",
+      useNewProtocol = useNewProtocol
+    )
+
+    for (version <- ApiKeys.DELETE_GROUPS.oldestVersion() to ApiKeys.DELETE_GROUPS.latestVersion(isUnstableApiEnabled)) {
+      // Join the consumer group. Note that we don't heartbeat here so we must use
+      // a session long enough for the duration of the test.
+      val (memberId, memberEpoch) = joinConsumerGroup(
+        groupId = "grp",
+        useNewProtocol = useNewProtocol
+      )
+
+      // The member leaves the group so that grp is empty and ready to be deleted.
+      leaveGroup(
+        groupId = "grp",
+        memberId = memberId,
+        useNewProtocol = useNewProtocol
+      )
+
+      deleteGroups(
+        groupIds = List("grp-non-empty", "grp"),
+        expectedErrors = List(Errors.NON_EMPTY_GROUP, Errors.NONE),
+        version = version.toShort
+      )
+
+      if (useNewProtocol) {
+        commitOffset(
+          groupId = "grp",
+          memberId = memberId,
+          memberEpoch = memberEpoch,
+          topic = "foo",
+          partition = 0,
+          offset = 100L,
+          expectedError = Errors.GROUP_ID_NOT_FOUND,
+          version = ApiKeys.OFFSET_COMMIT.latestVersion(isUnstableApiEnabled)
+        )
+      } else {
+        assertEquals(
+          List(new DescribedGroup()
+            .setGroupId("grp")
+            .setGroupState(GenericGroupState.DEAD.toString)
+          ),
+          describeGroups(
+            groupIds = List("grp"),
+            version = ApiKeys.DESCRIBE_GROUPS.latestVersion(isUnstableApiEnabled)
+          )
+        )
+      }
+    }
+  }
+}
diff --git a/core/src/test/scala/unit/kafka/server/DescribeGroupsRequestTest.scala b/core/src/test/scala/unit/kafka/server/DescribeGroupsRequestTest.scala
@@ -0,0 +1,119 @@
+/**
+ * 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 kafka.server
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.message.DescribeGroupsResponseData.{DescribedGroup, DescribedGroupMember}
+import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.coordinator.group.generic.GenericGroupState
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.{Tag, Timeout}
+import org.junit.jupiter.api.extension.ExtendWith
+
+import scala.jdk.CollectionConverters._
+
+@Timeout(120)
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(clusterType = Type.KRAFT, brokers = 1)
+@Tag("integration")
+class DescribeGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
+  @ClusterTest(serverProperties = Array(
+    new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+    new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+    new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+  ))
+  def testDescribeGroupsWithOldConsumerGroupProtocolAndNewGroupCoordinator(): Unit = {
+    testDescribeGroups()
+  }
+
+  @ClusterTest(clusterType = Type.ALL, serverProperties = Array(
+    new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "false"),
+    new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+    new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+  ))
+  def testDescribeGroupsWithOldConsumerGroupProtocolAndOldGroupCoordinator(): Unit = {
+    testDescribeGroups()
+  }
+
+  private def testDescribeGroups(): Unit = {
+    // Creates the __consumer_offsets topics because it won't be created automatically
+    // in this test because it does not use FindCoordinator API.
+    createOffsetsTopic()
+
+    // Create the topic.
+    createTopic(
+      topic = "foo",
+      numPartitions = 3
+    )
+
+    // Join the consumer group. Complete the rebalance so that grp-1 is in STABLE state.
+    val (memberId1, _) = joinConsumerGroupWithOldProtocol(
+      groupId = "grp-1",
+      metadata = Array(1, 2, 3),
+      assignment = Array(4, 5, 6)
+    )
+    // Join the consumer group. Not complete the rebalance so that grp-2 is in COMPLETING_REBALANCE state.
+    val (memberId2, _) = joinConsumerGroupWithOldProtocol(
+      groupId = "grp-2",
+      metadata = Array(1, 2, 3),
+      completeRebalance = false
+    )
+
+    for (version <- ApiKeys.DESCRIBE_GROUPS.oldestVersion() to ApiKeys.DESCRIBE_GROUPS.latestVersion(isUnstableApiEnabled)) {
+      assertEquals(
+        List(
+          new DescribedGroup()
+            .setGroupId("grp-1")
+            .setGroupState(GenericGroupState.STABLE.toString)
+            .setProtocolType("consumer")
+            .setProtocolData("consumer-range")
+            .setMembers(List(
+              new DescribedGroupMember()
+                .setMemberId(memberId1)
+                .setGroupInstanceId(null)
+                .setClientId("client-id")
+                .setClientHost("/127.0.0.1")
+                .setMemberMetadata(Array(1, 2, 3))
+                .setMemberAssignment(Array(4, 5, 6))
+            ).asJava),
+          new DescribedGroup()
+            .setGroupId("grp-2")
+            .setGroupState(GenericGroupState.COMPLETING_REBALANCE.toString)
+            .setProtocolType("consumer")
+            .setMembers(List(
+              new DescribedGroupMember()
+                .setMemberId(memberId2)
+                .setGroupInstanceId(null)
+                .setClientId("client-id")
+                .setClientHost("/127.0.0.1")
+                .setMemberMetadata(Array.empty)
+                .setMemberAssignment(Array.empty)
+            ).asJava),
+          new DescribedGroup()
+            .setGroupId("grp-unknown")
+            .setGroupState(GenericGroupState.DEAD.toString) // Return DEAD group when the group does not exist.
+        ),
+        describeGroups(
+          groupIds = List("grp-1", "grp-2", "grp-unknown"),
+          version = version.toShort
+        )
+      )
+    }
+  }
+}