KAFKA-12454: Add ERROR logging on kafka-log-dirs when given brokerIds do not exist in current kafka cluster

core/src/main/scala/kafka/admin/LogDirsCommand.scala

@@ -40,17 +40,27 @@ object LogDirsCommand {
         val opts = new LogDirsCommandOptions(args)
         val adminClient = createAdminClient(opts)
         val topicList = opts.options.valueOf(opts.topicListOpt).split(",").filter(!_.isEmpty)
+        val clusterBrokers: Array[Int] = adminClient.describeCluster().nodes().get().asScala.map(_.id()).toArray
+        var nonExistBrokers: Array[Int] = Array.emptyIntArray
         val brokerList = Option(opts.options.valueOf(opts.brokerListOpt)) match {
-            case Some(brokerListStr) => brokerListStr.split(',').filter(!_.isEmpty).map(_.toInt)
-            case None => adminClient.describeCluster().nodes().get().asScala.map(_.id()).toArray
+            case Some(brokerListStr) => {
+                val inputBrokers: Array[Int] = brokerListStr.split(',').filter(!_.isEmpty).map(_.toInt)
+                nonExistBrokers = inputBrokers.filterNot(brokerId => clusterBrokers.contains(brokerId))
+                inputBrokers
+            }
+            case None => clusterBrokers
         }
 
-        out.println("Querying brokers for log directories information")
-        val describeLogDirsResult: DescribeLogDirsResult = adminClient.describeLogDirs(brokerList.map(Integer.valueOf).toSeq.asJava)
-        val logDirInfosByBroker = describeLogDirsResult.allDescriptions.get().asScala.map { case (k, v) => k -> v.asScala }
+        if (!nonExistBrokers.isEmpty) {
+            out.println(s"ERROR: The given node(s) does not exist from broker-list ${nonExistBrokers.mkString(",")}")
+        } else {
+            out.println("Querying brokers for log directories information")
+            val describeLogDirsResult: DescribeLogDirsResult = adminClient.describeLogDirs(brokerList.map(Integer.valueOf).toSeq.asJava)
+            val logDirInfosByBroker = describeLogDirsResult.allDescriptions.get().asScala.map { case (k, v) => k -> v.asScala }
 
-        out.println(s"Received log directory information from brokers ${brokerList.mkString(",")}")
-        out.println(formatAsJson(logDirInfosByBroker, topicList.toSet))
+            out.println(s"Received log directory information from brokers ${brokerList.mkString(",")}")
+            out.println(formatAsJson(logDirInfosByBroker, topicList.toSet))
+        }
         adminClient.close()
     }
 

core/src/test/scala/unit/kafka/admin/LogDirsCommandTest.scala

@@ -0,0 +1,52 @@
+package unit.kafka.admin
+
+import java.io.{ByteArrayOutputStream, PrintStream}
+import java.nio.charset.StandardCharsets
+
+import kafka.admin.LogDirsCommand
+import kafka.integration.KafkaServerTestHarness
+import kafka.server.KafkaConfig
+import kafka.utils.TestUtils
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+import scala.collection.Seq
+
+class LogDirsCommandTest extends KafkaServerTestHarness {
+
+  def generateConfigs: Seq[KafkaConfig] = {
+    TestUtils.createBrokerConfigs(1, zkConnect)
+      .map(KafkaConfig.fromProps)
+  }
+
+  @Test
+  def checkLogDirsCommandOutput(): Unit = {
+    val byteArrayOutputStream = new ByteArrayOutputStream
+    val printStream = new PrintStream(byteArrayOutputStream, false, StandardCharsets.UTF_8.name())
+    //input exist brokerList
+    LogDirsCommand.describe(Array("--bootstrap-server", brokerList, "--broker-list", "0", "--describe"), printStream)
+    val existBrokersContent = new String(byteArrayOutputStream.toByteArray, StandardCharsets.UTF_8)
+    val existBrokersLineIter = existBrokersContent.split("\n").iterator
+
+    assertTrue(existBrokersLineIter.hasNext)
+    assertTrue(existBrokersLineIter.next().contains(s"Querying brokers for log directories information"))
+
+    //input nonExist brokerList
+    byteArrayOutputStream.reset()
+    LogDirsCommand.describe(Array("--bootstrap-server", brokerList, "--broker-list", "0,1,2", "--describe"), printStream)
+    val nonExistBrokersContent = new String(byteArrayOutputStream.toByteArray, StandardCharsets.UTF_8)
+    val nonExistBrokersLineIter = nonExistBrokersContent.split("\n").iterator
+
+    assertTrue(nonExistBrokersLineIter.hasNext)
+    assertTrue(nonExistBrokersLineIter.next().contains(s"ERROR: The given node(s) does not exist from broker-list 1,2"))
+
+    //use all brokerList for current cluster
+    byteArrayOutputStream.reset()
+    LogDirsCommand.describe(Array("--bootstrap-server", brokerList, "--describe"), printStream)
+    val allBrokersContent = new String(byteArrayOutputStream.toByteArray, StandardCharsets.UTF_8)
+    val allBrokersLineIter = allBrokersContent.split("\n").iterator
+
+    assertTrue(allBrokersLineIter.hasNext)
+    assertTrue(allBrokersLineIter.next().contains(s"Querying brokers for log directories information"))
+  }
+}