KAFKA-14589 ConsumerGroupCommand rewritten in java

bin/kafka-consumer-groups.sh

@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.ConsumerGroupCommand "$@"
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.consumergroup.ConsumerGroupCommand "$@"

bin/windows/kafka-consumer-groups.bat

@@ -14,4 +14,4 @@ rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 rem See the License for the specific language governing permissions and
 rem limitations under the License.
 
-"%~dp0kafka-run-class.bat" kafka.admin.ConsumerGroupCommand %*
+"%~dp0kafka-run-class.bat" org.apache.kafka.tools.consumergroup.ConsumerGroupCommand %*

build.gradle

@@ -1972,6 +1972,9 @@ project(':tools') {
     implementation project(':log4j-appender')
     implementation project(':tools:tools-api')
     implementation libs.argparse4j
+    implementation libs.jacksonDatabind
+    implementation libs.jacksonDataformatCsv
+    implementation libs.jacksonJDK8Datatypes
     implementation libs.slf4jApi
     implementation libs.log4j
     implementation libs.joptSimple
@@ -1986,6 +1989,8 @@ project(':tools') {
     testImplementation project(':server-common')
     testImplementation project(':server-common').sourceSets.test.output
     testImplementation project(':connect:api')
+    testImplementation project(':metadata')
+    testImplementation project(':metadata').sourceSets.test.output
     testImplementation project(':connect:runtime')
     testImplementation project(':connect:runtime').sourceSets.test.output
     testImplementation project(':storage:storage-api').sourceSets.main.output

checkstyle/import-control.xml

@@ -325,6 +325,12 @@
       <allow pkg="scala" />
     </subpackage>
 
+    <subpackage name="consumergroup">
+      <allow pkg="kafka.security"/>
+      <allow pkg="org.apache.kafka.metadata.authorizer"/>
+      <allow pkg="org.apache.kafka.tools"/>
+    </subpackage>
+
     <subpackage name="other">
       <allow pkg="org.apache.kafka.tools.reassign"/>
       <allow pkg="kafka.log" />

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -593,6 +593,16 @@ public static <T> String join(T[] strs, String separator) {
         return join(Arrays.asList(strs), separator);
     }
 
+    /**
+     * Create a string representation of a collection joined by ", ".
+     * @param collection The list of items
+     * @return The string representation.
+     */
+    public static <T> String join(Collection<T> collection) {
+        Objects.requireNonNull(collection);
+        return mkString(collection.stream(), "", "", ", ");
+    }
+
     /**
      * Create a string representation of a collection joined by the given separator
      * @param collection The list of items

core/src/main/scala/kafka/utils/ToolsUtils.scala

@@ -69,7 +69,7 @@ object ToolsUtils {
   /**
    * This is a simple wrapper around `CommandLineUtils.printUsageAndExit`.
    * It is needed for tools migration (KAFKA-14525), as there is no Java equivalent for return type `Nothing`.
-   * Can be removed once [[kafka.admin.ConsumerGroupCommand]], [[kafka.tools.ConsoleConsumer]]
+   * Can be removed once [[kafka.tools.ConsoleConsumer]]
    * and [[kafka.tools.ConsoleProducer]] are migrated.
    *
    * @param parser Command line options parser.

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -18,7 +18,6 @@ import java.util
 import java.util.concurrent.ExecutionException
 import java.util.regex.Pattern
 import java.util.{Collections, Optional, Properties}
-import kafka.admin.ConsumerGroupCommand.{ConsumerGroupCommandOptions, ConsumerGroupService}
 import kafka.security.authorizer.{AclAuthorizer, AclEntry}
 import kafka.security.authorizer.AclEntry.WildcardHost
 import kafka.server.{BaseRequestTest, KafkaConfig}
@@ -1709,20 +1708,6 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
     createAdminClient().describeConsumerGroups(Seq(group).asJava).describedGroups().get(group).get()
   }
 
-  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
-  @ValueSource(strings = Array("zk", "kraft"))
-  def testDescribeGroupCliWithGroupDescribe(quorum: String): Unit = {
-    createTopicWithBrokerPrincipal(topic)
-    addAndVerifyAcls(Set(new AccessControlEntry(clientPrincipalString, WildcardHost, DESCRIBE, ALLOW)), groupResource)
-    addAndVerifyAcls(Set(new AccessControlEntry(clientPrincipalString, WildcardHost, DESCRIBE, ALLOW)), topicResource)
-
-    val cgcArgs = Array("--bootstrap-server", bootstrapServers(), "--describe", "--group", group)
-    val opts = new ConsumerGroupCommandOptions(cgcArgs)
-    val consumerGroupService = new ConsumerGroupService(opts)
-    consumerGroupService.describeGroups()
-    consumerGroupService.close()
-  }
-
   @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
   @ValueSource(strings = Array("zk", "kraft"))
   def testListGroupApiWithAndWithoutListGroupAcls(quorum: String): Unit = {

core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala

@@ -23,7 +23,6 @@ import org.apache.kafka.common.{KafkaException, TopicPartition}
 import org.apache.kafka.common.errors.SaslAuthenticationException
 import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo}
 import org.junit.jupiter.api.Assertions._
-import kafka.admin.ConsumerGroupCommand.{ConsumerGroupCommandOptions, ConsumerGroupService}
 import kafka.server.KafkaConfig
 import kafka.utils.{JaasTestUtils, TestUtils}
 import kafka.zk.ConfigEntityChangeNotificationZNode
@@ -32,7 +31,7 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.ValueSource
 
 class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with SaslSetup {
-  private val kafkaClientSaslMechanism = "SCRAM-SHA-256"
+  val kafkaClientSaslMechanism = "SCRAM-SHA-256"
   private val kafkaServerSaslMechanisms = List(kafkaClientSaslMechanism)
   override protected val securityProtocol = SecurityProtocol.SASL_PLAINTEXT
   override protected val serverSaslProperties = Some(kafkaServerSaslProperties(kafkaServerSaslMechanisms, kafkaClientSaslMechanism))
@@ -166,46 +165,7 @@ class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with
     }
   }
 
-  @Test
-  def testConsumerGroupServiceWithAuthenticationFailure(): Unit = {
-    val consumerGroupService: ConsumerGroupService = prepareConsumerGroupService
-
-    val consumer = createConsumer()
-    try {
-      consumer.subscribe(List(topic).asJava)
-
-      verifyAuthenticationException(consumerGroupService.listGroups())
-    } finally consumerGroupService.close()
-  }
-
-  @Test
-  def testConsumerGroupServiceWithAuthenticationSuccess(): Unit = {
-    createClientCredential()
-    val consumerGroupService: ConsumerGroupService = prepareConsumerGroupService
-
-    val consumer = createConsumer()
-    try {
-      consumer.subscribe(List(topic).asJava)
-
-      verifyWithRetry(consumer.poll(Duration.ofMillis(1000)))
-      assertEquals(1, consumerGroupService.listConsumerGroups().size)
-    }
-    finally consumerGroupService.close()
-  }
-
-  private def prepareConsumerGroupService = {
-    val propsFile = TestUtils.tempPropertiesFile(Map("security.protocol" -> "SASL_PLAINTEXT", "sasl.mechanism" -> kafkaClientSaslMechanism))
-
-    val cgcArgs = Array("--bootstrap-server", bootstrapServers(),
-                        "--describe",
-                        "--group", "test.group",
-                        "--command-config", propsFile.getAbsolutePath)
-    val opts = new ConsumerGroupCommandOptions(cgcArgs)
-    val consumerGroupService = new ConsumerGroupService(opts)
-    consumerGroupService
-  }
-
-  private def createClientCredential(): Unit = {
+  def createClientCredential(): Unit = {
     createScramCredentialsViaPrivilegedAdminClient(JaasTestUtils.KafkaScramUser2, JaasTestUtils.KafkaScramPassword2)
   }
 
@@ -221,14 +181,14 @@ class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with
     }
   }
 
-  private def verifyAuthenticationException(action: => Unit): Unit = {
+  def verifyAuthenticationException(action: => Unit): Unit = {
     val startMs = System.currentTimeMillis
     assertThrows(classOf[Exception], () => action)
     val elapsedMs = System.currentTimeMillis - startMs
     assertTrue(elapsedMs <= 5000, s"Poll took too long, elapsed=$elapsedMs")
   }
 
-  private def verifyWithRetry(action: => Unit): Unit = {
+  def verifyWithRetry(action: => Unit): Unit = {
     var attempts = 0
     TestUtils.waitUntilTrue(() => {
       try {