[GitHub] [kafka] wenbingshen commented on a change in pull request #10304: KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds do not exist in current kafka cluster

[GitBox <gi...@apache.org>]

wenbingshen commented on a change in pull request #10304:
URL: https://github.com/apache/kafka/pull/10304#discussion_r593706745



##########
File path: core/src/main/scala/kafka/admin/LogDirsCommand.scala
##########
@@ -40,9 +40,16 @@ 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
         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 None => clusterBrokers
+        }
+
+        val nonExistBrokers: Array[Int] = brokerList.filterNot(brokerId => clusterBrokers.contains(brokerId))
+        if (!nonExistBrokers.isEmpty) {
+          System.err.println(s"The given node(s) does not exist from broker-list ${nonExistBrokers.mkString(",")}")
+          sys.exit(1)

Review comment:
       > Should we do this only when handle the brokers provided by the user? It does not make sense to validate the list of brokers otherwise. What do you think?
   
   Your suggestion is very good.I have changed the logic, please review.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org