You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/03/14 15:19:14 UTC
[GitHub] [kafka] chia7712 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
chia7712 commented on a change in pull request #10304:
URL: https://github.com/apache/kafka/pull/10304#discussion_r593915131
##########
File path: 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) => {
Review comment:
the `{}` is redundant.
##########
File path: 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) {
Review comment:
maybe `nonEmpty`?
##########
File path: 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)
Review comment:
Could you replace `!_.isEmpty` by `_.nonEmpty`? It seems to me the later is more readable.
##########
File path: 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))
Review comment:
How about using `diff` to get `nonExistBrokers`?
##########
File path: 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
Review comment:
Is `Set[Int]` more suitable for this case?
##########
File path: 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(",")}")
Review comment:
Could we show existent brokers also?
##########
File path: 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()
Review comment:
Could you use `try-finally` to release `adminClient`?
----------------------------------------------------------------
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