You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2024/02/13 03:02:43 UTC
(kafka) branch trunk updated: KAFKA-14589: [1/3] Tests of ConsoleGroupCommand rewritten in java (#15256)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 88c5543ccfb KAFKA-14589: [1/3] Tests of ConsoleGroupCommand rewritten in java (#15256)
88c5543ccfb is described below
commit 88c5543ccfb517ceba8dc56d22576efdf2540e0e
Author: Nikolay <ni...@apache.org>
AuthorDate: Tue Feb 13 06:02:36 2024 +0300
KAFKA-14589: [1/3] Tests of ConsoleGroupCommand rewritten in java (#15256)
This PR is part of #14471
Is contains some of ConsoleGroupCommand tests rewritten in java.
Intention of separate PR is to reduce changes and simplify review.
Reviewers: Luke Chen <sh...@gmail.com>
---
build.gradle | 2 +
checkstyle/import-control.xml | 2 +
.../scala/kafka/admin/ConsumerGroupCommand.scala | 2 +-
.../kafka/admin/DeleteConsumerGroupsTest.scala | 258 ---------------
...ffsetsConsumerGroupCommandIntegrationTest.scala | 209 ------------
.../scala/unit/kafka/admin/DeleteTopicTest.scala | 2 +
.../unit/kafka/admin/ListConsumerGroupTest.scala | 147 ---------
.../apache/kafka/tools/{reassign => }/Tuple2.java | 2 +-
.../tools/reassign/ReassignPartitionsCommand.java | 1 +
.../consumer/group/ConsumerGroupCommandTest.java | 359 +++++++++++++++++++++
.../consumer/group/DeleteConsumerGroupsTest.java | 310 ++++++++++++++++++
...OffsetsConsumerGroupCommandIntegrationTest.java | 216 +++++++++++++
.../consumer/group/ListConsumerGroupTest.java | 164 ++++++++++
.../ReassignPartitionsIntegrationTest.java | 1 +
.../tools/reassign/ReassignPartitionsUnitTest.java | 1 +
15 files changed, 1060 insertions(+), 616 deletions(-)
diff --git a/build.gradle b/build.gradle
index d3652334651..e715144d54f 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2012,6 +2012,8 @@ project(':tools') {
testImplementation project(':clients')
testImplementation project(':clients').sourceSets.test.output
+ testImplementation project(':server')
+ testImplementation project(':server').sourceSets.test.output
testImplementation project(':core')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':server-common')
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index b6fa73b68ca..2f9c3960483 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -332,7 +332,9 @@
</subpackage>
<subpackage name="consumer.group">
+ <allow pkg="kafka.api"/>
<allow pkg="org.apache.kafka.tools"/>
+ <allow pkg="org.apache.kafka.server.config" />
<allow pkg="scala"/>
</subpackage>
diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
index 26c5be3b981..4187274a22d 100755
--- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
@@ -135,7 +135,7 @@ object ConsumerGroupCommand extends Logging {
private[admin] case class MemberAssignmentState(group: String, consumerId: String, host: String, clientId: String, groupInstanceId: String,
numPartitions: Int, assignment: List[TopicPartition])
- private[admin] case class GroupState(group: String, coordinator: Node, assignmentStrategy: String, state: String, numMembers: Int)
+ case class GroupState(group: String, coordinator: Node, assignmentStrategy: String, state: String, numMembers: Int)
private[admin] sealed trait CsvRecord
private[admin] case class CsvRecordWithGroup(group: String, topic: String, partition: Int, offset: Long) extends CsvRecord
diff --git a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala
deleted file mode 100644
index a5bdda2d025..00000000000
--- a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala
+++ /dev/null
@@ -1,258 +0,0 @@
-/**
- * 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.admin
-
-import joptsimple.OptionException
-import kafka.utils.{TestInfoUtils, TestUtils}
-import org.apache.kafka.common.errors.{GroupIdNotFoundException, GroupNotEmptyException}
-import org.apache.kafka.common.protocol.Errors
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
-
-class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteWithTopicOption(quorum: String): Unit = {
- createOffsetsTopic()
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", group, "--topic")
- assertThrows(classOf[OptionException], () => getConsumerGroupService(cgcArgs))
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteCmdNonExistingGroup(quorum: String): Unit = {
- createOffsetsTopic()
- val missingGroup = "missing.group"
-
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", missingGroup)
- val service = getConsumerGroupService(cgcArgs)
-
- val output = TestUtils.grabConsoleOutput(service.deleteGroups())
- assertTrue(output.contains(s"Group '$missingGroup' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message),
- s"The expected error (${Errors.GROUP_ID_NOT_FOUND}) was not detected while deleting consumer group")
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteNonExistingGroup(quorum: String): Unit = {
- createOffsetsTopic()
- val missingGroup = "missing.group"
-
- // note the group to be deleted is a different (non-existing) group
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", missingGroup)
- val service = getConsumerGroupService(cgcArgs)
-
- val result = service.deleteGroups()
- assertTrue(result.size == 1 && result.keySet.contains(missingGroup) && result(missingGroup).getCause.isInstanceOf[GroupIdNotFoundException],
- s"The expected error (${Errors.GROUP_ID_NOT_FOUND}) was not detected while deleting consumer group")
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteCmdNonEmptyGroup(quorum: String): Unit = {
- createOffsetsTopic()
-
- // run one consumer in the group
- addConsumerGroupExecutor(numConsumers = 1)
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", group)
- val service = getConsumerGroupService(cgcArgs)
-
- TestUtils.waitUntilTrue(() => {
- service.collectGroupMembers(group, false)._2.get.size == 1
- }, "The group did not initialize as expected.")
-
- val output = TestUtils.grabConsoleOutput(service.deleteGroups())
- assertTrue(output.contains(s"Group '$group' could not be deleted due to:") && output.contains(Errors.NON_EMPTY_GROUP.message),
- s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group. Output was: (${output})")
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteNonEmptyGroup(quorum: String): Unit = {
- createOffsetsTopic()
-
- // run one consumer in the group
- addConsumerGroupExecutor(numConsumers = 1)
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", group)
- val service = getConsumerGroupService(cgcArgs)
-
- TestUtils.waitUntilTrue(() => {
- service.collectGroupMembers(group, false)._2.get.size == 1
- }, "The group did not initialize as expected.")
-
- val result = service.deleteGroups()
- assertNotNull(result(group),
- s"Group was deleted successfully, but it shouldn't have been. Result was:(${result})")
- assertTrue(result.size == 1 && result.keySet.contains(group) && result(group).getCause.isInstanceOf[GroupNotEmptyException],
- s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group. Result was:(${result})")
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteCmdEmptyGroup(quorum: String): Unit = {
- createOffsetsTopic()
-
- // run one consumer in the group
- val executor = addConsumerGroupExecutor(numConsumers = 1)
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", group)
- val service = getConsumerGroupService(cgcArgs)
-
- TestUtils.waitUntilTrue(() => {
- service.listConsumerGroups().contains(group) && service.collectGroupState(group).state == "Stable"
- }, "The group did not initialize as expected.")
-
- executor.shutdown()
-
- TestUtils.waitUntilTrue(() => {
- service.collectGroupState(group).state == "Empty"
- }, "The group did not become empty as expected.")
-
- val output = TestUtils.grabConsoleOutput(service.deleteGroups())
- assertTrue(output.contains(s"Deletion of requested consumer groups ('$group') was successful."),
- s"The consumer group could not be deleted as expected")
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteCmdAllGroups(quorum: String): Unit = {
- createOffsetsTopic()
-
- // Create 3 groups with 1 consumer per each
- val groups =
- (for (i <- 1 to 3) yield {
- val group = this.group + i
- val executor = addConsumerGroupExecutor(numConsumers = 1, group = group)
- group -> executor
- }).toMap
-
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--all-groups")
- val service = getConsumerGroupService(cgcArgs)
-
- TestUtils.waitUntilTrue(() => {
- service.listConsumerGroups().toSet == groups.keySet &&
- groups.keySet.forall(groupId => service.collectGroupState(groupId).state == "Stable")
- }, "The group did not initialize as expected.")
-
- // Shutdown consumers to empty out groups
- groups.values.foreach(executor => executor.shutdown())
-
- TestUtils.waitUntilTrue(() => {
- groups.keySet.forall(groupId => service.collectGroupState(groupId).state == "Empty")
- }, "The group did not become empty as expected.")
-
- val output = TestUtils.grabConsoleOutput(service.deleteGroups()).trim
- val expectedGroupsForDeletion = groups.keySet
- val deletedGroupsGrepped = output.substring(output.indexOf('(') + 1, output.indexOf(')')).split(',')
- .map(_.replaceAll("'", "").trim).toSet
-
- assertTrue(output.matches(s"Deletion of requested consumer groups (.*) was successful.")
- && deletedGroupsGrepped == expectedGroupsForDeletion, s"The consumer group(s) could not be deleted as expected"
- )
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteEmptyGroup(quorum: String): Unit = {
- createOffsetsTopic()
-
- // run one consumer in the group
- val executor = addConsumerGroupExecutor(numConsumers = 1)
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", group)
- val service = getConsumerGroupService(cgcArgs)
-
- TestUtils.waitUntilTrue(() => {
- service.listConsumerGroups().contains(group) && service.collectGroupState(group).state == "Stable"
- }, "The group did not initialize as expected.")
-
- executor.shutdown()
-
- TestUtils.waitUntilTrue(() => {
- service.collectGroupState(group).state == "Empty"
- }, "The group did not become empty as expected.")
-
- val result = service.deleteGroups()
- assertTrue(result.size == 1 && result.keySet.contains(group) && result(group) == null,
- s"The consumer group could not be deleted as expected")
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteCmdWithMixOfSuccessAndError(quorum: String): Unit = {
- createOffsetsTopic()
- val missingGroup = "missing.group"
-
- // run one consumer in the group
- val executor = addConsumerGroupExecutor(numConsumers = 1)
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", group)
- val service = getConsumerGroupService(cgcArgs)
-
- TestUtils.waitUntilTrue(() => {
- service.listConsumerGroups().contains(group) && service.collectGroupState(group).state == "Stable"
- }, "The group did not initialize as expected.")
-
- executor.shutdown()
-
- TestUtils.waitUntilTrue(() => {
- service.collectGroupState(group).state == "Empty"
- }, "The group did not become empty as expected.")
-
- val service2 = getConsumerGroupService(cgcArgs ++ Array("--group", missingGroup))
- val output = TestUtils.grabConsoleOutput(service2.deleteGroups())
- assertTrue(output.contains(s"Group '$missingGroup' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message) &&
- output.contains(s"These consumer groups were deleted successfully: '$group'"), s"The consumer group deletion did not work as expected")
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteWithMixOfSuccessAndError(quorum: String): Unit = {
- createOffsetsTopic()
- val missingGroup = "missing.group"
-
- // run one consumer in the group
- val executor = addConsumerGroupExecutor(numConsumers = 1)
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--delete", "--group", group)
- val service = getConsumerGroupService(cgcArgs)
-
- TestUtils.waitUntilTrue(() => {
- service.listConsumerGroups().contains(group) && service.collectGroupState(group).state == "Stable"
- }, "The group did not initialize as expected.")
-
- executor.shutdown()
-
- TestUtils.waitUntilTrue(() => {
- service.collectGroupState(group).state == "Empty"
- }, "The group did not become empty as expected.")
-
- val service2 = getConsumerGroupService(cgcArgs ++ Array("--group", missingGroup))
- val result = service2.deleteGroups()
- assertTrue(result.size == 2 &&
- result.keySet.contains(group) && result(group) == null &&
- result.keySet.contains(missingGroup) &&
- result(missingGroup).getMessage.contains(Errors.GROUP_ID_NOT_FOUND.message),
- s"The consumer group deletion did not work as expected")
- }
-
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteWithUnrecognizedNewConsumerOption(quorum: String): Unit = {
- val cgcArgs = Array("--new-consumer", "--bootstrap-server", bootstrapServers(), "--delete", "--group", group)
- assertThrows(classOf[OptionException], () => getConsumerGroupService(cgcArgs))
- }
-}
diff --git a/core/src/test/scala/unit/kafka/admin/DeleteOffsetsConsumerGroupCommandIntegrationTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteOffsetsConsumerGroupCommandIntegrationTest.scala
deleted file mode 100644
index e7e73a260e1..00000000000
--- a/core/src/test/scala/unit/kafka/admin/DeleteOffsetsConsumerGroupCommandIntegrationTest.scala
+++ /dev/null
@@ -1,209 +0,0 @@
-/**
- * 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.admin
-
-import java.util.Properties
-import kafka.utils.TestUtils
-import kafka.utils.TestInfoUtils
-import org.apache.kafka.clients.consumer.Consumer
-import org.apache.kafka.clients.consumer.ConsumerConfig
-import org.apache.kafka.clients.consumer.KafkaConsumer
-import org.apache.kafka.clients.producer.KafkaProducer
-import org.apache.kafka.clients.producer.ProducerConfig
-import org.apache.kafka.clients.producer.ProducerRecord
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.serialization.ByteArrayDeserializer
-import org.apache.kafka.common.serialization.ByteArraySerializer
-import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.server.config.Defaults
-
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
-
-class DeleteOffsetsConsumerGroupCommandIntegrationTest extends ConsumerGroupCommandTest {
-
- def getArgs(group: String, topic: String): Array[String] = {
- Array(
- "--bootstrap-server", bootstrapServers(),
- "--delete-offsets",
- "--group", group,
- "--topic", topic
- )
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsNonExistingGroup(quorum: String): Unit = {
- val group = "missing.group"
- val topic = "foo:1"
- val service = getConsumerGroupService(getArgs(group, topic))
-
- val (error, _) = service.deleteOffsets(group, List(topic))
- assertEquals(Errors.GROUP_ID_NOT_FOUND, error)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfStableConsumerGroupWithTopicPartition(quorum: String): Unit = {
- testWithStableConsumerGroup(topic, 0, 0, Errors.GROUP_SUBSCRIBED_TO_TOPIC)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfStableConsumerGroupWithTopicOnly(quorum: String): Unit = {
- testWithStableConsumerGroup(topic, -1, 0, Errors.GROUP_SUBSCRIBED_TO_TOPIC)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfStableConsumerGroupWithUnknownTopicPartition(quorum: String): Unit = {
- testWithStableConsumerGroup("foobar", 0, 0, Errors.UNKNOWN_TOPIC_OR_PARTITION)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfStableConsumerGroupWithUnknownTopicOnly(quorum: String): Unit = {
- testWithStableConsumerGroup("foobar", -1, -1, Errors.UNKNOWN_TOPIC_OR_PARTITION)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfEmptyConsumerGroupWithTopicPartition(quorum: String): Unit = {
- testWithEmptyConsumerGroup(topic, 0, 0, Errors.NONE)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfEmptyConsumerGroupWithTopicOnly(quorum: String): Unit = {
- testWithEmptyConsumerGroup(topic, -1, 0, Errors.NONE)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfEmptyConsumerGroupWithUnknownTopicPartition(quorum: String): Unit = {
- testWithEmptyConsumerGroup("foobar", 0, 0, Errors.UNKNOWN_TOPIC_OR_PARTITION)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteOffsetsOfEmptyConsumerGroupWithUnknownTopicOnly(quorum: String): Unit = {
- testWithEmptyConsumerGroup("foobar", -1, -1, Errors.UNKNOWN_TOPIC_OR_PARTITION)
- }
-
- private def testWithStableConsumerGroup(inputTopic: String,
- inputPartition: Int,
- expectedPartition: Int,
- expectedError: Errors): Unit = {
- testWithConsumerGroup(
- withStableConsumerGroup,
- inputTopic,
- inputPartition,
- expectedPartition,
- expectedError)
- }
-
- private def testWithEmptyConsumerGroup(inputTopic: String,
- inputPartition: Int,
- expectedPartition: Int,
- expectedError: Errors): Unit = {
- testWithConsumerGroup(
- withEmptyConsumerGroup,
- inputTopic,
- inputPartition,
- expectedPartition,
- expectedError)
- }
-
- private def testWithConsumerGroup(withConsumerGroup: (=> Unit) => Unit,
- inputTopic: String,
- inputPartition: Int,
- expectedPartition: Int,
- expectedError: Errors): Unit = {
- produceRecord()
- withConsumerGroup {
- val topic = if (inputPartition >= 0) inputTopic + ":" + inputPartition else inputTopic
- val service = getConsumerGroupService(getArgs(group, topic))
- val (topLevelError, partitions) = service.deleteOffsets(group, List(topic))
- val tp = new TopicPartition(inputTopic, expectedPartition)
- // Partition level error should propagate to top level, unless this is due to a missed partition attempt.
- if (inputPartition >= 0) {
- assertEquals(expectedError, topLevelError)
- }
- if (expectedError == Errors.NONE)
- assertNull(partitions(tp))
- else
- assertEquals(expectedError.exception, partitions(tp).getCause)
- }
- }
-
- private def produceRecord(): Unit = {
- val producer = createProducer()
- try {
- producer.send(new ProducerRecord(topic, 0, null, null)).get()
- } finally {
- Utils.closeQuietly(producer, "producer")
- }
- }
-
- private def withStableConsumerGroup(body: => Unit): Unit = {
- val consumer = createConsumer()
- try {
- TestUtils.subscribeAndWaitForRecords(this.topic, consumer)
- consumer.commitSync()
- body
- } finally {
- Utils.closeQuietly(consumer, "consumer")
- }
- }
-
- private def withEmptyConsumerGroup(body: => Unit): Unit = {
- val consumer = createConsumer()
- try {
- TestUtils.subscribeAndWaitForRecords(this.topic, consumer)
- consumer.commitSync()
- } finally {
- Utils.closeQuietly(consumer, "consumer")
- }
- body
- }
-
- private def createProducer(config: Properties = new Properties()): KafkaProducer[Array[Byte], Array[Byte]] = {
- config.putIfAbsent(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
- config.putIfAbsent(ProducerConfig.ACKS_CONFIG, "-1")
- config.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[ByteArraySerializer].getName)
- config.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[ByteArraySerializer].getName)
-
- new KafkaProducer(config)
- }
-
- private def createConsumer(config: Properties = new Properties()): Consumer[Array[Byte], Array[Byte]] = {
- config.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
- config.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
- config.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, group)
- config.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[ByteArrayDeserializer].getName)
- config.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[ByteArrayDeserializer].getName)
- // Increase timeouts to avoid having a rebalance during the test
- config.putIfAbsent(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, Integer.MAX_VALUE.toString)
- config.putIfAbsent(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Defaults.GROUP_MAX_SESSION_TIMEOUT_MS.toString)
-
- new KafkaConsumer(config)
- }
-
-}
diff --git a/core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala
index e02266e0a48..894a47d5608 100644
--- a/core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala
@@ -488,6 +488,8 @@ class DeleteTopicTest extends QuorumTestHarness {
// Restart KRaft quorum with the updated config
val overridingProps = new Properties()
overridingProps.put(KafkaConfig.DeleteTopicEnableProp, false.toString)
+ if (implementation != null)
+ implementation.shutdown()
implementation = newKRaftQuorum(overridingProps)
}
diff --git a/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala b/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala
deleted file mode 100644
index 1aa56d51297..00000000000
--- a/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala
+++ /dev/null
@@ -1,147 +0,0 @@
-/**
- * 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.admin
-
-import joptsimple.OptionException
-import org.junit.jupiter.api.Assertions._
-import kafka.utils.TestUtils
-import org.apache.kafka.common.ConsumerGroupState
-import org.apache.kafka.clients.admin.ConsumerGroupListing
-import java.util.Optional
-
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
-
-class ListConsumerGroupTest extends ConsumerGroupCommandTest {
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testListConsumerGroups(quorum: String): Unit = {
- val simpleGroup = "simple-group"
- addSimpleGroupExecutor(group = simpleGroup)
- addConsumerGroupExecutor(numConsumers = 1)
-
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--list")
- val service = getConsumerGroupService(cgcArgs)
-
- val expectedGroups = Set(group, simpleGroup)
- var foundGroups = Set.empty[String]
- TestUtils.waitUntilTrue(() => {
- foundGroups = service.listConsumerGroups().toSet
- expectedGroups == foundGroups
- }, s"Expected --list to show groups $expectedGroups, but found $foundGroups.")
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testListWithUnrecognizedNewConsumerOption(): Unit = {
- val cgcArgs = Array("--new-consumer", "--bootstrap-server", bootstrapServers(), "--list")
- assertThrows(classOf[OptionException], () => getConsumerGroupService(cgcArgs))
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testListConsumerGroupsWithStates(): Unit = {
- val simpleGroup = "simple-group"
- addSimpleGroupExecutor(group = simpleGroup)
- addConsumerGroupExecutor(numConsumers = 1)
-
- val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--list", "--state")
- val service = getConsumerGroupService(cgcArgs)
-
- val expectedListing = Set(
- new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
- new ConsumerGroupListing(group, false, Optional.of(ConsumerGroupState.STABLE)))
-
- var foundListing = Set.empty[ConsumerGroupListing]
- TestUtils.waitUntilTrue(() => {
- foundListing = service.listConsumerGroupsWithState(ConsumerGroupState.values.toSet).toSet
- expectedListing == foundListing
- }, s"Expected to show groups $expectedListing, but found $foundListing")
-
- val expectedListingStable = Set(
- new ConsumerGroupListing(group, false, Optional.of(ConsumerGroupState.STABLE)))
-
- foundListing = Set.empty[ConsumerGroupListing]
- TestUtils.waitUntilTrue(() => {
- foundListing = service.listConsumerGroupsWithState(Set(ConsumerGroupState.STABLE)).toSet
- expectedListingStable == foundListing
- }, s"Expected to show groups $expectedListingStable, but found $foundListing")
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testConsumerGroupStatesFromString(quorum: String): Unit = {
- var result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable")
- assertEquals(Set(ConsumerGroupState.STABLE), result)
-
- result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance")
- assertEquals(Set(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE), result)
-
- result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,")
- assertEquals(Set(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE), result)
-
- result = ConsumerGroupCommand.consumerGroupStatesFromString("stable")
- assertEquals(Set(ConsumerGroupState.STABLE), result)
-
- result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning")
- assertEquals(Set(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING), result)
-
- result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,")
- assertEquals(Set(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING), result)
-
- assertThrows(classOf[IllegalArgumentException], () => ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"))
-
- assertThrows(classOf[IllegalArgumentException], () => ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"))
-
- assertThrows(classOf[IllegalArgumentException], () => ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"))
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testListGroupCommand(quorum: String): Unit = {
- val simpleGroup = "simple-group"
- addSimpleGroupExecutor(group = simpleGroup)
- addConsumerGroupExecutor(numConsumers = 1)
- var out = ""
-
- var cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--list")
- TestUtils.waitUntilTrue(() => {
- out = TestUtils.grabConsoleOutput(ConsumerGroupCommand.main(cgcArgs))
- !out.contains("STATE") && out.contains(simpleGroup) && out.contains(group)
- }, s"Expected to find $simpleGroup, $group and no header, but found $out")
-
- cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--list", "--state")
- TestUtils.waitUntilTrue(() => {
- out = TestUtils.grabConsoleOutput(ConsumerGroupCommand.main(cgcArgs))
- out.contains("STATE") && out.contains(simpleGroup) && out.contains(group)
- }, s"Expected to find $simpleGroup, $group and the header, but found $out")
-
- cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--list", "--state", "Stable")
- TestUtils.waitUntilTrue(() => {
- out = TestUtils.grabConsoleOutput(ConsumerGroupCommand.main(cgcArgs))
- out.contains("STATE") && out.contains(group) && out.contains("Stable")
- }, s"Expected to find $group in state Stable and the header, but found $out")
-
- cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--list", "--state", "stable")
- TestUtils.waitUntilTrue(() => {
- out = TestUtils.grabConsoleOutput(ConsumerGroupCommand.main(cgcArgs))
- out.contains("STATE") && out.contains(group) && out.contains("Stable")
- }, s"Expected to find $group in state Stable and the header, but found $out")
- }
-
-}
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/Tuple2.java b/tools/src/main/java/org/apache/kafka/tools/Tuple2.java
similarity index 97%
rename from tools/src/main/java/org/apache/kafka/tools/reassign/Tuple2.java
rename to tools/src/main/java/org/apache/kafka/tools/Tuple2.java
index a9e84317cf7..02dd4bf3981 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/Tuple2.java
+++ b/tools/src/main/java/org/apache/kafka/tools/Tuple2.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.tools.reassign;
+package org.apache.kafka.tools;
import java.util.Objects;
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
index 6a2d9b93468..36233469807 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
@@ -49,6 +49,7 @@ import org.apache.kafka.server.util.json.JsonObject;
import org.apache.kafka.server.util.json.JsonValue;
import org.apache.kafka.tools.TerseException;
import org.apache.kafka.tools.ToolsUtils;
+import org.apache.kafka.tools.Tuple2;
import java.io.IOException;
import java.util.ArrayList;
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTest.java
new file mode 100644
index 00000000000..b78054cb4ad
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTest.java
@@ -0,0 +1,359 @@
+/*
+ * 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.tools.consumer.group;
+
+import kafka.api.BaseConsumerTest;
+import kafka.admin.ConsumerGroupCommand;
+import kafka.server.KafkaConfig;
+import kafka.utils.TestUtils;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.params.provider.Arguments;
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+public class ConsumerGroupCommandTest extends kafka.integration.KafkaServerTestHarness {
+ public static final String TOPIC = "foo";
+ public static final String GROUP = "test.group";
+
+ List<ConsumerGroupCommand.ConsumerGroupService> consumerGroupService = new ArrayList<>();
+ List<AbstractConsumerGroupExecutor> consumerGroupExecutors = new ArrayList<>();
+
+ @Override
+ public Seq<kafka.server.KafkaConfig> generateConfigs() {
+ List<KafkaConfig> cfgs = new ArrayList<>();
+
+ TestUtils.createBrokerConfigs(
+ 1,
+ zkConnectOrNull(),
+ false,
+ true,
+ scala.None$.empty(),
+ scala.None$.empty(),
+ scala.None$.empty(),
+ true,
+ false,
+ false,
+ false,
+ scala.collection.immutable.Map$.MODULE$.empty(),
+ 1,
+ false,
+ 1,
+ (short) 1,
+ 0,
+ false
+ ).foreach(props -> {
+ if (isNewGroupCoordinatorEnabled()) {
+ props.setProperty(KafkaConfig.NewGroupCoordinatorEnableProp(), "true");
+ }
+
+ cfgs.add(KafkaConfig.fromProps(props));
+ return null;
+ });
+
+ return seq(cfgs);
+ }
+
+ @BeforeEach
+ @Override
+ public void setUp(TestInfo testInfo) {
+ super.setUp(testInfo);
+ createTopic(TOPIC, 1, 1, new Properties(), listenerName(), new Properties());
+ }
+
+ @AfterEach
+ @Override
+ public void tearDown() {
+ consumerGroupService.forEach(ConsumerGroupCommand.ConsumerGroupService::close);
+ consumerGroupExecutors.forEach(AbstractConsumerGroupExecutor::shutdown);
+ super.tearDown();
+ }
+
+ Map<TopicPartition, Long> committedOffsets(String topic, String group) {
+ try (Consumer<String, String> consumer = createNoAutoCommitConsumer(group)) {
+ Set<TopicPartition> partitions = consumer.partitionsFor(topic).stream()
+ .map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()))
+ .collect(Collectors.toSet());
+ return consumer.committed(partitions).entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));
+ }
+ }
+
+ Consumer<String, String> createNoAutoCommitConsumer(String group) {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", bootstrapServers(listenerName()));
+ props.put("group.id", group);
+ props.put("enable.auto.commit", "false");
+ return new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer());
+ }
+
+ ConsumerGroupCommand.ConsumerGroupService getConsumerGroupService(String[] args) {
+ ConsumerGroupCommand.ConsumerGroupCommandOptions opts = new ConsumerGroupCommand.ConsumerGroupCommandOptions(args);
+ ConsumerGroupCommand.ConsumerGroupService service = new ConsumerGroupCommand.ConsumerGroupService(
+ opts,
+ asScala(Collections.singletonMap(AdminClientConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE)))
+ );
+
+ consumerGroupService.add(0, service);
+ return service;
+ }
+
+ ConsumerGroupExecutor addConsumerGroupExecutor(int numConsumers) {
+ return addConsumerGroupExecutor(numConsumers, TOPIC, GROUP, RangeAssignor.class.getName(), Optional.empty(), Optional.empty(), false, GroupProtocol.CLASSIC.name);
+ }
+
+ ConsumerGroupExecutor addConsumerGroupExecutor(int numConsumers, String groupProtocol) {
+ return addConsumerGroupExecutor(numConsumers, TOPIC, GROUP, RangeAssignor.class.getName(), Optional.empty(), Optional.empty(), false, groupProtocol);
+ }
+
+ ConsumerGroupExecutor addConsumerGroupExecutor(int numConsumers, String groupProtocol, Optional<String> remoteAssignor) {
+ return addConsumerGroupExecutor(numConsumers, TOPIC, GROUP, RangeAssignor.class.getName(), remoteAssignor, Optional.empty(), false, groupProtocol);
+ }
+
+ ConsumerGroupExecutor addConsumerGroupExecutor(int numConsumers, String topic, String group) {
+ return addConsumerGroupExecutor(numConsumers, topic, group, RangeAssignor.class.getName(), Optional.empty(), Optional.empty(), false, GroupProtocol.CLASSIC.name);
+ }
+
+ ConsumerGroupExecutor addConsumerGroupExecutor(int numConsumers, String topic, String group, String groupProtocol) {
+ return addConsumerGroupExecutor(numConsumers, topic, group, RangeAssignor.class.getName(), Optional.empty(), Optional.empty(), false, groupProtocol);
+ }
+
+ ConsumerGroupExecutor addConsumerGroupExecutor(int numConsumers, String topic, String group, String strategy, Optional<String> remoteAssignor,
+ Optional<Properties> customPropsOpt, boolean syncCommit, String groupProtocol) {
+ ConsumerGroupExecutor executor = new ConsumerGroupExecutor(bootstrapServers(listenerName()), numConsumers, group, groupProtocol,
+ topic, strategy, remoteAssignor, customPropsOpt, syncCommit);
+ addExecutor(executor);
+ return executor;
+ }
+
+ SimpleConsumerGroupExecutor addSimpleGroupExecutor(String group) {
+ return addSimpleGroupExecutor(Arrays.asList(new TopicPartition(TOPIC, 0)), group);
+ }
+
+ SimpleConsumerGroupExecutor addSimpleGroupExecutor(Collection<TopicPartition> partitions, String group) {
+ SimpleConsumerGroupExecutor executor = new SimpleConsumerGroupExecutor(bootstrapServers(listenerName()), group, partitions);
+ addExecutor(executor);
+ return executor;
+ }
+
+ private AbstractConsumerGroupExecutor addExecutor(AbstractConsumerGroupExecutor executor) {
+ consumerGroupExecutors.add(0, executor);
+ return executor;
+ }
+
+ abstract class AbstractConsumerRunnable implements Runnable {
+ final String broker;
+ final String groupId;
+ final Optional<Properties> customPropsOpt;
+ final boolean syncCommit;
+
+ final Properties props = new Properties();
+ KafkaConsumer<String, String> consumer;
+
+ boolean configured = false;
+
+ public AbstractConsumerRunnable(String broker, String groupId, Optional<Properties> customPropsOpt, boolean syncCommit) {
+ this.broker = broker;
+ this.groupId = groupId;
+ this.customPropsOpt = customPropsOpt;
+ this.syncCommit = syncCommit;
+ }
+
+ void configure() {
+ configured = true;
+ configure(props);
+ customPropsOpt.ifPresent(props::putAll);
+ consumer = new KafkaConsumer<>(props);
+ }
+
+ void configure(Properties props) {
+ props.put("bootstrap.servers", broker);
+ props.put("group.id", groupId);
+ props.put("key.deserializer", StringDeserializer.class.getName());
+ props.put("value.deserializer", StringDeserializer.class.getName());
+ }
+
+ abstract void subscribe();
+
+ @Override
+ public void run() {
+ assert configured : "Must call configure before use";
+ try {
+ subscribe();
+ while (true) {
+ consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
+ if (syncCommit)
+ consumer.commitSync();
+ }
+ } catch (WakeupException e) {
+ // OK
+ } finally {
+ consumer.close();
+ }
+ }
+
+ void shutdown() {
+ consumer.wakeup();
+ }
+ }
+
+ class ConsumerRunnable extends AbstractConsumerRunnable {
+ final String topic;
+ final String groupProtocol;
+ final String strategy;
+ final Optional<String> remoteAssignor;
+
+ public ConsumerRunnable(String broker, String groupId, String groupProtocol, String topic, String strategy,
+ Optional<String> remoteAssignor, Optional<Properties> customPropsOpt, boolean syncCommit) {
+ super(broker, groupId, customPropsOpt, syncCommit);
+
+ this.topic = topic;
+ this.groupProtocol = groupProtocol;
+ this.strategy = strategy;
+ this.remoteAssignor = remoteAssignor;
+ }
+
+ @Override
+ void configure(Properties props) {
+ super.configure(props);
+ props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol);
+ if (groupProtocol.toUpperCase(Locale.ROOT).equals(GroupProtocol.CONSUMER.toString())) {
+ remoteAssignor.ifPresent(assignor -> props.put(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, assignor));
+ } else {
+ props.put("partition.assignment.strategy", strategy);
+ }
+ }
+
+ @Override
+ void subscribe() {
+ consumer.subscribe(Collections.singleton(topic));
+ }
+ }
+
+ class SimpleConsumerRunnable extends AbstractConsumerRunnable {
+ final Collection<TopicPartition> partitions;
+
+ public SimpleConsumerRunnable(String broker, String groupId, Collection<TopicPartition> partitions) {
+ super(broker, groupId, Optional.empty(), false);
+
+ this.partitions = partitions;
+ }
+
+ @Override
+ void subscribe() {
+ consumer.assign(partitions);
+ }
+ }
+
+ class AbstractConsumerGroupExecutor {
+ final int numThreads;
+ final ExecutorService executor;
+ final List<AbstractConsumerRunnable> consumers = new ArrayList<>();
+
+ public AbstractConsumerGroupExecutor(int numThreads) {
+ this.numThreads = numThreads;
+ this.executor = Executors.newFixedThreadPool(numThreads);
+ }
+
+ void submit(AbstractConsumerRunnable consumerThread) {
+ consumers.add(consumerThread);
+ executor.submit(consumerThread);
+ }
+
+ void shutdown() {
+ consumers.forEach(AbstractConsumerRunnable::shutdown);
+ executor.shutdown();
+ try {
+ executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ class ConsumerGroupExecutor extends AbstractConsumerGroupExecutor {
+ public ConsumerGroupExecutor(String broker, int numConsumers, String groupId, String groupProtocol, String topic, String strategy,
+ Optional<String> remoteAssignor, Optional<Properties> customPropsOpt, boolean syncCommit) {
+ super(numConsumers);
+ IntStream.rangeClosed(1, numConsumers).forEach(i -> {
+ ConsumerRunnable th = new ConsumerRunnable(broker, groupId, groupProtocol, topic, strategy, remoteAssignor, customPropsOpt, syncCommit);
+ th.configure();
+ submit(th);
+ });
+ }
+ }
+
+ class SimpleConsumerGroupExecutor extends AbstractConsumerGroupExecutor {
+ public SimpleConsumerGroupExecutor(String broker, String groupId, Collection<TopicPartition> partitions) {
+ super(1);
+
+ SimpleConsumerRunnable th = new SimpleConsumerRunnable(broker, groupId, partitions);
+ th.configure();
+ submit(th);
+ }
+ }
+
+
+ public static Stream<Arguments> getTestQuorumAndGroupProtocolParametersAll() {
+ return BaseConsumerTest.getTestQuorumAndGroupProtocolParametersAll();
+ }
+
+ @SuppressWarnings({"deprecation"})
+ static <T> Seq<T> seq(Collection<T> seq) {
+ return JavaConverters.asScalaIteratorConverter(seq.iterator()).asScala().toSeq();
+ }
+
+ @SuppressWarnings("deprecation")
+ static <K, V> scala.collection.Map<K, V> asScala(Map<K, V> jmap) {
+ return JavaConverters.mapAsScalaMap(jmap);
+ }
+
+ @SuppressWarnings({"deprecation"})
+ static <T> scala.collection.immutable.Set<T> set(final Collection<T> set) {
+ return JavaConverters.asScalaSet(new HashSet<>(set)).toSet();
+ }
+}
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java
new file mode 100644
index 00000000000..0a1bcb2daef
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java
@@ -0,0 +1,310 @@
+/*
+ * 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.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithTopicOption(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--topic"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // note the group to be deleted is a different (non-existing) group
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(missingGroup) && result.get(missingGroup).get().getCause() instanceof GroupIdNotFoundException,
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + GROUP + "' could not be deleted due to:") && output.contains(Errors.NON_EMPTY_GROUP.message()),
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Output was: (" + output + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertNotNull(result.get(GROUP).get(),
+ "Group was deleted successfully, but it shouldn't have been. Result was:(" + result + ")");
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get().getCause() instanceof GroupNotEmptyException,
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Result was:(" + result + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected."
+ );
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Deletion of requested consumer groups ('" + GROUP + "') was successful."),
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdAllGroups(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // Create 3 groups with 1 consumer per each
+ Map<String, ConsumerGroupExecutor> groups = IntStream.rangeClosed(1, 3).mapToObj(i -> GROUP + i).collect(Collectors.toMap(
+ Function.identity(),
+ group -> addConsumerGroupExecutor(1, TOPIC, group, RangeAssignor.class.getName(), Optional.empty(), Optional.empty(), false, GroupProtocol.CLASSIC.name)
+ ));
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--all-groups"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(() ->
+ Objects.equals(service.listConsumerGroups().toSet(), set(groups.keySet())) &&
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Stable");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not initialize as expected.");
+
+ // Shutdown consumers to empty out groups
+ groups.values().forEach(AbstractConsumerGroupExecutor::shutdown);
+
+ TestUtils.waitForCondition(() ->
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Empty");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not become empty as expected.");
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ }).trim();
+ Set<String> expectedGroupsForDeletion = groups.keySet();
+ Set<String> deletedGroupsGrepped = Arrays.stream(output.substring(output.indexOf('(') + 1, output.indexOf(')')).split(","))
+ .map(str -> str.replaceAll("'", "").trim()).collect(Collectors.toSet());
+
+ assertTrue(output.matches("Deletion of requested consumer groups (.*) was successful.")
+ && Objects.equals(deletedGroupsGrepped, expectedGroupsForDeletion),
+ "The consumer group(s) could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get() == null,
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--group", missingGroup};
+
+ ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service2.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:")
+ && output.contains(Errors.GROUP_ID_NOT_FOUND.message())
+ && output.contains("These consumer groups were deleted successfully: '" + GROUP + "'"),
+ "The consumer group deletion did not work as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--group", missingGroup};
+
+ ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs);
+ scala.collection.Map<String, Throwable> result = service2.deleteGroups();
+ assertTrue(result.size() == 2 &&
+ result.contains(GROUP) && result.get(GROUP).get() == null &&
+ result.contains(missingGroup) &&
+ result.get(missingGroup).get().getMessage().contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The consumer group deletion did not work as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithUnrecognizedNewConsumerOption(String quorum) {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+}
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteOffsetsConsumerGroupCommandIntegrationTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteOffsetsConsumerGroupCommandIntegrationTest.java
new file mode 100644
index 00000000000..2ac093923ce
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteOffsetsConsumerGroupCommandIntegrationTest.java
@@ -0,0 +1,216 @@
+/*
+ * 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.tools.consumer.group;
+
+import kafka.admin.ConsumerGroupCommand;
+import kafka.utils.TestUtils;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.config.Defaults;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class DeleteOffsetsConsumerGroupCommandIntegrationTest extends ConsumerGroupCommandTest {
+ String[] getArgs(String group, String topic) {
+ return new String[] {
+ "--bootstrap-server", bootstrapServers(listenerName()),
+ "--delete-offsets",
+ "--group", group,
+ "--topic", topic
+ };
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsNonExistingGroup() {
+ String group = "missing.group";
+ String topic = "foo:1";
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(getArgs(group, topic));
+
+ scala.Tuple2<Errors, scala.collection.Map<TopicPartition, Throwable>> res = service.deleteOffsets(group, seq(Collections.singleton(topic)).toList());
+ assertEquals(Errors.GROUP_ID_NOT_FOUND, res._1);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfStableConsumerGroupWithTopicPartition() {
+ testWithStableConsumerGroup(TOPIC, 0, 0, Errors.GROUP_SUBSCRIBED_TO_TOPIC);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfStableConsumerGroupWithTopicOnly() {
+ testWithStableConsumerGroup(TOPIC, -1, 0, Errors.GROUP_SUBSCRIBED_TO_TOPIC);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfStableConsumerGroupWithUnknownTopicPartition() {
+ testWithStableConsumerGroup("foobar", 0, 0, Errors.UNKNOWN_TOPIC_OR_PARTITION);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfStableConsumerGroupWithUnknownTopicOnly() {
+ testWithStableConsumerGroup("foobar", -1, -1, Errors.UNKNOWN_TOPIC_OR_PARTITION);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfEmptyConsumerGroupWithTopicPartition() {
+ testWithEmptyConsumerGroup(TOPIC, 0, 0, Errors.NONE);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfEmptyConsumerGroupWithTopicOnly() {
+ testWithEmptyConsumerGroup(TOPIC, -1, 0, Errors.NONE);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfEmptyConsumerGroupWithUnknownTopicPartition() {
+ testWithEmptyConsumerGroup("foobar", 0, 0, Errors.UNKNOWN_TOPIC_OR_PARTITION);
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteOffsetsOfEmptyConsumerGroupWithUnknownTopicOnly() {
+ testWithEmptyConsumerGroup("foobar", -1, -1, Errors.UNKNOWN_TOPIC_OR_PARTITION);
+ }
+
+ private void testWithStableConsumerGroup(String inputTopic,
+ int inputPartition,
+ int expectedPartition,
+ Errors expectedError) {
+ testWithConsumerGroup(
+ this::withStableConsumerGroup,
+ inputTopic,
+ inputPartition,
+ expectedPartition,
+ expectedError);
+ }
+
+ private void testWithEmptyConsumerGroup(String inputTopic,
+ int inputPartition,
+ int expectedPartition,
+ Errors expectedError) {
+ testWithConsumerGroup(
+ this::withEmptyConsumerGroup,
+ inputTopic,
+ inputPartition,
+ expectedPartition,
+ expectedError);
+ }
+
+ private void testWithConsumerGroup(java.util.function.Consumer<Runnable> withConsumerGroup,
+ String inputTopic,
+ int inputPartition,
+ int expectedPartition,
+ Errors expectedError) {
+ produceRecord();
+ withConsumerGroup.accept(() -> {
+ String topic = inputPartition >= 0 ? inputTopic + ":" + inputPartition : inputTopic;
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(getArgs(GROUP, topic));
+ scala.Tuple2<Errors, scala.collection.Map<TopicPartition, Throwable>> res = service.deleteOffsets(GROUP, seq(Collections.singletonList(topic)).toList());
+ Errors topLevelError = res._1;
+ scala.collection.Map<TopicPartition, Throwable> partitions = res._2;
+ TopicPartition tp = new TopicPartition(inputTopic, expectedPartition);
+ // Partition level error should propagate to top level, unless this is due to a missed partition attempt.
+ if (inputPartition >= 0) {
+ assertEquals(expectedError, topLevelError);
+ }
+ if (expectedError == Errors.NONE)
+ assertNull(partitions.get(tp).get());
+ else
+ assertEquals(expectedError.exception(), partitions.get(tp).get().getCause());
+ });
+ }
+
+ private void produceRecord() {
+ KafkaProducer<byte[], byte[]> producer = createProducer(new Properties());
+ try {
+ producer.send(new ProducerRecord<>(TOPIC, 0, null, null)).get();
+ } catch (ExecutionException | InterruptedException e) {
+ throw new RuntimeException(e);
+ } finally {
+ Utils.closeQuietly(producer, "producer");
+ }
+ }
+
+ private void withStableConsumerGroup(Runnable body) {
+ Consumer<byte[], byte[]> consumer = createConsumer(new Properties());
+ try {
+ TestUtils.subscribeAndWaitForRecords(TOPIC, consumer, DEFAULT_MAX_WAIT_MS);
+ consumer.commitSync();
+ body.run();
+ } finally {
+ Utils.closeQuietly(consumer, "consumer");
+ }
+ }
+
+ private void withEmptyConsumerGroup(Runnable body) {
+ Consumer<byte[], byte[]> consumer = createConsumer(new Properties());
+ try {
+ TestUtils.subscribeAndWaitForRecords(TOPIC, consumer, DEFAULT_MAX_WAIT_MS);
+ consumer.commitSync();
+ } finally {
+ Utils.closeQuietly(consumer, "consumer");
+ }
+ body.run();
+ }
+
+ private KafkaProducer<byte[], byte[]> createProducer(Properties config) {
+ config.putIfAbsent(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers(listenerName()));
+ config.putIfAbsent(ProducerConfig.ACKS_CONFIG, "-1");
+ config.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+ config.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+
+ return new KafkaProducer<>(config);
+ }
+
+ private Consumer<byte[], byte[]> createConsumer(Properties config) {
+ config.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers(listenerName()));
+ config.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ config.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, GROUP);
+ config.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+ config.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+ // Increase timeouts to avoid having a rebalance during the test
+ config.putIfAbsent(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
+ config.putIfAbsent(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.toString(Defaults.GROUP_MAX_SESSION_TIMEOUT_MS));
+
+ return new KafkaConsumer<>(config);
+ }
+}
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java
new file mode 100644
index 00000000000..6648cfe2a24
--- /dev/null
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final AtomicReference<scala.collection.Set> foundGroups = new AtomicReference<>();
+ TestUtils.waitForCondition(() -> {
+ foundGroups.set(service.listConsumerGroups().toSet());
+ return Objects.equals(expectedGroups, foundGroups.get());
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups.get() + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final AtomicReference<scala.collection.Set> foundListing = new AtomicReference<>();
+ TestUtils.waitForCondition(() -> {
+ foundListing.set(service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet());
+ return Objects.equals(expectedListing, foundListing.get());
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing.get());
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing.set(null);
+
+ TestUtils.waitForCondition(() -> {
+ foundListing.set(service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet());
+ return Objects.equals(expectedListingStable, foundListing.get());
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListGroupCommand(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+ final AtomicReference<String> out = new AtomicReference<>("");
+
+ String[] cgcArgs1 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs1);
+ return null;
+ }));
+ return !out.get().contains("STATE") && out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + simpleGroup + ", " + GROUP + " and no header, but found " + out.get());
+
+ String[] cgcArgs2 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs2);
+ return null;
+ }));
+ return out.get().contains("STATE") && out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + simpleGroup + ", " + GROUP + " and the header, but found " + out.get());
+
+ String[] cgcArgs3 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "Stable"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs3);
+ return null;
+ }));
+ return out.get().contains("STATE") && out.get().contains(GROUP) && out.get().contains("Stable");
+ }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out.get());
+
+ String[] cgcArgs4 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "stable"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ }));
+ return out.get().contains("STATE") && out.get().contains(GROUP) && out.get().contains("Stable");
+ }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out.get());
+ }
+}
diff --git a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java
index b71331a6b23..83fc665e3e4 100644
--- a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java
@@ -43,6 +43,7 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.tools.TerseException;
+import org.apache.kafka.tools.Tuple2;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
diff --git a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java
index d0e1cb4baaf..699e048fb20 100644
--- a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java
@@ -31,6 +31,7 @@ import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.common.AdminCommandFailedException;
import org.apache.kafka.server.common.AdminOperationException;
+import org.apache.kafka.tools.Tuple2;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;