[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_r595727918
##
File path: core/src/test/scala/unit/kafka/admin/LogDirsCommandTest.scala
##
@@ -0,0 +1,77 @@
+/**
+ * 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 unit.kafka.admin
Review comment:
the package name should be `kafka.admin` rather than `package
unit.kafka.admin`
##
File path: core/src/main/scala/kafka/admin/LogDirsCommand.scala
##
@@ -39,19 +39,30 @@ object LogDirsCommand {
def describe(args: Array[String], out: PrintStream): Unit = {
val opts = new LogDirsCommandOptions(args)
val adminClient = createAdminClient(opts)
-val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(!_.isEmpty)
-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
-}
+try {
+val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(_.nonEmpty)
+val clusterBrokers =
adminClient.describeCluster().nodes().get().asScala.map(_.id()).toSet
+val (existingBrokers, nonExistingBrokers) =
Option(opts.options.valueOf(opts.brokerListOpt)) match {
+case Some(brokerListStr) =>
+val inputBrokers =
brokerListStr.split(',').filter(_.nonEmpty).map(_.toInt).toSet
+(inputBrokers, inputBrokers.diff(clusterBrokers))
Review comment:
As the variable is called `existingBrokers `, we should find out the
"true" existent brokers. In short, it should return
`inputBrokers.intersect(clusterBrokers)` rather than `inputBrokers`
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:
us...@infra.apache.org
[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_r595008454
##
File path: core/src/main/scala/kafka/admin/LogDirsCommand.scala
##
@@ -39,19 +39,29 @@ object LogDirsCommand {
def describe(args: Array[String], out: PrintStream): Unit = {
val opts = new LogDirsCommandOptions(args)
val adminClient = createAdminClient(opts)
-val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(!_.isEmpty)
-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
-}
+val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(_.nonEmpty)
+try {
+val clusterBrokers =
adminClient.describeCluster().nodes().get().asScala.map(_.id()).toSet
+val (existingBrokers, nonExistingBrokers) =
Option(opts.options.valueOf(opts.brokerListOpt)) match {
+case Some(brokerListStr) =>
+val inputBrokers =
brokerListStr.split(',').filter(_.nonEmpty).map(_.toInt).toSet
+(inputBrokers, inputBrokers.diff(clusterBrokers))
+case None => (clusterBrokers, Set.empty)
+}
-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 (nonExistingBrokers.nonEmpty) {
+out.println(s"ERROR: The given brokers do not exist from
--broker-list: ${nonExistingBrokers.mkString(",")}. Current cluster exist
brokers: ${clusterBrokers.mkString(",")}")
Review comment:
How about `current existent brokers: `
##
File path: core/src/main/scala/kafka/admin/LogDirsCommand.scala
##
@@ -39,19 +39,29 @@ object LogDirsCommand {
def describe(args: Array[String], out: PrintStream): Unit = {
val opts = new LogDirsCommandOptions(args)
val adminClient = createAdminClient(opts)
-val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(!_.isEmpty)
-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
-}
+val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(_.nonEmpty)
Review comment:
the resource declaration should be followed by `try` block.
```scala
val adminClient = createAdminClient(opts)
try {
}
```
##
File path: core/src/main/scala/kafka/admin/LogDirsCommand.scala
##
@@ -39,19 +39,29 @@ object LogDirsCommand {
def describe(args: Array[String], out: PrintStream): Unit = {
val opts = new LogDirsCommandOptions(args)
val adminClient = createAdminClient(opts)
-val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(!_.isEmpty)
-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
-}
+val topicList =
opts.options.valueOf(opts.topicListOpt).split(",").filter(_.nonEmpty)
+try {
+val clusterBrokers =
adminClient.describeCluster().nodes().get().asScala.map(_.id()).toSet
+val (existingBrokers, nonExistingBrokers) =
Option(opts.options.valueOf(opts.brokerListOpt)) match {
+case Some(brokerListStr) =>
+val inputBrokers =
brokerListStr.split(',').filter(_.nonEmpty).map(_.toInt).toSet
+(inputBrokers, inputBrokers.diff(clusterBrokers))
+case None => (clusterBrokers, Set.empty)
+}
-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 (nonExistingBrokers.nonEmpty) {
+out.println(s"ERROR: The given brokers do not exist from
--broker-list: ${nonExistingBrokers.mkString(",")}. Current cluster exist
brokers: ${clusterBrokers.mkString(",")}")
Review comment:
Also, could you separate this line?
[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
