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

wenbingshen · 2021-03-11T17:38:30Z

When non-existent brokerIds value are given, the kafka-log-dirs tool will have a timeout error:

Exception in thread "main" java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeLogDirs
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
at kafka.admin.LogDirsCommand$.describe(LogDirsCommand.scala:50)
at kafka.admin.LogDirsCommand$.main(LogDirsCommand.scala:36)
at kafka.admin.LogDirsCommand.main(LogDirsCommand.scala)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeLogDirs

When the brokerId entered by the user does not exist, an error message indicating that the node is not present should be printed.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…do not exist in current kafka cluster

wenbingshen · 2021-03-11T17:40:10Z

@chia7712 Can you help review this PR? Thank you.

dajac

Thanks for the PR. I left a minor suggestion. Could we also add a test case?

dajac · 2021-03-12T23:20:37Z

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

+        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)


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?

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.

…do not exist in current kafka cluster

wenbingshen · 2021-03-13T07:37:25Z

Thanks for the PR. I left a minor suggestion. Could we also add a test case?

Thanks for your comment.Your suggestion is very good, we only need to judge on the node entered by the user, I have added a unit test, please review it again.Thank you.

chia7712

@wenbingshen thanks for this contribution. a couple of comments are left. Most of them are related to code style. Otherwise, LGTM

chia7712 · 2021-03-14T15:07:32Z

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

-            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)


Could you replace !_.isEmpty by _.nonEmpty? It seems to me the later is more readable.

chia7712 · 2021-03-14T15:10:17Z

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

        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) => {


the {} is redundant.

chia7712 · 2021-03-14T15:12:22Z

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

-        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) {


maybe nonEmpty?

chia7712 · 2021-03-14T15:12:56Z

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


Is Set[Int] more suitable for this case?

chia7712 · 2021-03-14T15:14:38Z

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

-            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))


How about using diff to get nonExistBrokers?

chia7712 · 2021-03-14T15:15:09Z

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

-        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()


Could you use try-finally to release adminClient?

chia7712 · 2021-03-14T15:17:41Z

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

-        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(",")}")


Could we show existent brokers also?

…do not exist in current kafka cluster

wenbingshen · 2021-03-14T16:30:50Z

@wenbingshen thanks for this contribution. a couple of comments are left. Most of them are related to code style. Otherwise, LGTM

Thank you very much for your patience and guidance. These comments are very important. I submitted them after modifying them. Please review them again, thank you.

dajac

@wenbingshen Thanks for the updates. I have left few more minor comments. Also, it seems that the build failed. Could you check it?

dajac · 2021-03-15T09:42:37Z

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

+        var nonExistBrokers: Set[Int] = Set.empty
+        try {
+            val clusterBrokers: Set[Int] = adminClient.describeCluster().nodes().get().asScala.map(_.id()).toSet
+            val brokerList = Option(opts.options.valueOf(opts.brokerListOpt)) match {
+                case Some(brokerListStr) =>
+                    val inputBrokers: Set[Int] = brokerListStr.split(',').filter(_.nonEmpty).map(_.toInt).toSet
+                    nonExistBrokers = inputBrokers.diff(clusterBrokers)
+                    inputBrokers
+                case None => clusterBrokers
+            }


nit: We usually avoid using mutable variable unless it is really necessary. In this case, I would rather return the nonExistingBrokers when the argument is processed. Something like this:

val (existingBrokers, nonExistingBrokers) = Option(opts.options.valueOf(opts.brokerListOpt)) match { case Some(brokerListStr) => val inputBrokers: Set[Int] = brokerListStr.split(',').filter(_.nonEmpty).map(_.toInt).toSet (inputBrokers, inputBrokers.diff(clusterBrokers) case None => (clusterBrokers, Set.empty) }

dajac · 2021-03-15T09:43:45Z

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

+        val topicList = opts.options.valueOf(opts.topicListOpt).split(",").filter(_.nonEmpty)
+        var nonExistBrokers: Set[Int] = Set.empty
+        try {
+            val clusterBrokers: Set[Int] = adminClient.describeCluster().nodes().get().asScala.map(_.id()).toSet


nit: We can remove specifying Set[Int].

dajac · 2021-03-15T09:44:01Z

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

+            val clusterBrokers: Set[Int] = adminClient.describeCluster().nodes().get().asScala.map(_.id()).toSet
+            val brokerList = Option(opts.options.valueOf(opts.brokerListOpt)) match {
+                case Some(brokerListStr) =>
+                    val inputBrokers: Set[Int] = brokerListStr.split(',').filter(_.nonEmpty).map(_.toInt).toSet


dajac · 2021-03-15T09:44:17Z

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

+                out.println(s"ERROR: The given node(s) does not exist from broker-list: ${nonExistBrokers.mkString(",")}. Current cluster exist node(s): ${clusterBrokers.mkString(",")}")
+            } else {
+                out.println("Querying brokers for log directories information")
+                val describeLogDirsResult: DescribeLogDirsResult = adminClient.describeLogDirs(brokerList.map(Integer.valueOf).toSeq.asJava)


nit: We can remove DescribeLogDirsResult.

…do not exist in current kafka cluster

wenbingshen · 2021-03-15T15:11:23Z

@wenbingshen Thanks for the updates. I have left few more minor comments. Also, it seems that the build failed. Could you check it?

Thank you for your review and suggestions. I have submitted the latest code, and the code has been tested and compiled successfully. Please help review it again, thank you.

…do not exist in current kafka cluster

dajac

@wenbingshen Thanks for the updates. Let few more minot comments.

dajac · 2021-03-15T16:11:51Z

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

@@ -0,0 +1,52 @@
+package unit.kafka.admin


We must add the licence header here. You can copy it from another file.

Sorry, after checking the compilation report, I have realized this problem and I have made changes.

dajac · 2021-03-15T16:13:01Z

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

-        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 node(s) does not exist from broker-list: ${nonExistingBrokers.mkString(",")}. Current cluster exist node(s): ${clusterBrokers.mkString(",")}")


nit: Should we say --broker-list instead of broker-list? Also, should we say broker(s) instead of node(s) to be consistent with the message below?

Good idea.I will act right away.

dajac · 2021-03-15T16:13:15Z

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

+                out.println(s"ERROR: The given node(s) does not exist from broker-list: ${nonExistingBrokers.mkString(",")}. Current cluster exist node(s): ${clusterBrokers.mkString(",")}")
+            } else {
+                out.println("Querying brokers for log directories information")
+                val describeLogDirsResult: DescribeLogDirsResult = adminClient.describeLogDirs(existingBrokers.map(Integer.valueOf).toSeq.asJava)


nit: DescribeLogDirsResult can be removed.

Sorry, I forgot this, I will change it right away

…do not exist in current kafka cluster

wenbingshen · 2021-03-15T17:04:04Z

@wenbingshen Thanks for the updates. Let few more minot comments.

Thank you for your commonts.I submitted the latest code, please review it, thank you!

…do not exist in current kafka cluster

chia7712

@wenbingshen thanks for your patch. overall LGTM. a couple of trivial comments are left. please take a look :)

chia7712 · 2021-03-16T09:46:25Z

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

-        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(",")}")


How about current existent brokers:

chia7712 · 2021-03-16T09:46:52Z

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

-        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(",")}")


Also, could you separate this line?

chia7712 · 2021-03-16T09:49:23Z

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

-            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)


the resource declaration should be followed by try block.

val adminClient = createAdminClient(opts) try { }

Good.It has been modified in the latest submission.

chia7712 · 2021-03-16T09:55:13Z

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

+    assertTrue(existBrokersLineIter.hasNext)
+    assertTrue(existBrokersLineIter.next().contains(s"Querying brokers for log directories information"))
+
+    //input nonExist brokerList


how about using nonexistent instead of nonExist?

It has been modified.

chia7712 · 2021-03-16T09:56:12Z

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

+
+    //input nonExist brokerList
+    byteArrayOutputStream.reset()
+    LogDirsCommand.describe(Array("--bootstrap-server", brokerList, "--broker-list", "0,1,2", "--describe"), printStream)


Could you input duplicate ids and the check the output does not include duplicates?

@chia7712 I have added the duplicate ids test and it passed.

…do not exist in current kafka cluster

wenbingshen · 2021-03-16T10:59:49Z

@wenbingshen thanks for your patch. overall LGTM. a couple of trivial comments are left. please take a look :)

Thanks for your comments.According to your comments, I have submitted the latest code, please review it again :)

wenbingshen · 2021-03-17T05:27:41Z

Good afternoon @chia7712 @dajac Dear committers, if you have any comments on this PR, I will continue to improve it. If you are satisfied with it, can I apply for this PR to be merged into the trunk? :)

chia7712

@wenbingshen thanks for your patch. a couple of comments. please take a look.

chia7712 · 2021-03-17T05:45:28Z

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

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package unit.kafka.admin


the package name should be kafka.admin rather than package unit.kafka.admin

chia7712 · 2021-03-17T05:49:36Z

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

+            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))


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

…do not exist in current kafka cluster

wenbingshen · 2021-03-17T06:31:51Z

@wenbingshen thanks for your patch. a couple of comments. please take a look.

Thanks for your comments.I submitted the latest code, please review it again. :)

wenbingshen · 2021-03-18T03:33:52Z

Good afternoon @chia7712 @dajac Dear committers, if you have any comments on this PR, I will continue to improve it. If you are satisfied with it, can I apply for this PR to be merged into the trunk? :)

Dear @chia7712 @dajac If there are no other problems,can you help advance this pr?Thanks very much! :)

dajac

LGTM. @chia7712 If you are fine with the PR, I let you merge it.

chia7712 · 2021-03-18T07:37:35Z

LGTM. @chia7712 If you are fine with the PR, I let you merge it.

sure. will merge it after I take a final review :)

Conflicts: * build.gradle: keep `dependencySubstitution` Confluent addition in `resolutionStrategy` and take upstream changes. Commits: * apache-github/trunk: KAFKA-12503: inform threads to resize their cache instead of doing so for them (apache#10356) KAFKA-10697: Remove ProduceResponse.responses (apache#10332) MINOR: Exclude KIP-500.md from rat check (apache#10354) MINOR: Move `configurations.all` to be a child of `allprojects` (apache#10349) MINOR: Remove use of `NoSuchElementException` in `KafkaMetadataLog` (apache#10344) MINOR: Start the broker-to-controller channel for request forwarding (apache#10340) KAFKA-12382: add a README for KIP-500 (apache#10227) MINOR: Fix BaseHashTable sizing (apache#10334) KAFKA-10357: Add setup method to internal topics (apache#10317) MINOR: remove redundant null check when testing specified type (apache#10314) KAFKA-12293: Remove JCenter from buildscript and delete buildscript.gradle KAFKA-12491: Make rocksdb an `api` dependency for `streams` (apache#10341) KAFKA-12454: Add ERROR logging on kafka-log-dirs when given brokerIds do not exist in current kafka cluster (apache#10304) KAFKA-12459; Use property testing library for raft event simulation tests (apache#10323) MINOR: fix failing ZooKeeper system tests (apache#10297) MINOR: fix client_compatibility_features_test.py (apache#10292)

wenbingshen added 2 commits March 12, 2021 01:31

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

e71e5a4

…do not exist in current kafka cluster

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

91fd305

…do not exist in current kafka cluster

dajac reviewed Mar 12, 2021

View reviewed changes

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

b473a44

…do not exist in current kafka cluster

wenbingshen requested a review from dajac March 13, 2021 07:47

chia7712 reviewed Mar 14, 2021

View reviewed changes

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

a4d6ae7

…do not exist in current kafka cluster

wenbingshen requested a review from chia7712 March 14, 2021 16:32

dajac reviewed Mar 15, 2021

View reviewed changes

shenwenbing added 2 commits March 15, 2021 21:49

Merge branch 'trunk' into addErrorLog2LogDirsCommand

2b623e2

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

fefdef9

…do not exist in current kafka cluster

wenbingshen requested a review from dajac March 15, 2021 15:02

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

c88b348

…do not exist in current kafka cluster

dajac reviewed Mar 15, 2021

View reviewed changes

wenbingshen added 2 commits March 16, 2021 00:24

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

848822f

…do not exist in current kafka cluster

Merge remote-tracking branch 'sri/trunk' into KAFKA-12454

587800d

wenbingshen requested a review from dajac March 15, 2021 17:04

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

1e26832

…do not exist in current kafka cluster

chia7712 reviewed Mar 16, 2021

View reviewed changes

shenwenbing added 2 commits March 16, 2021 18:28

Merge branch 'trunk' into addErrorLog2LogDirsCommand

5641815

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

b245e5f

…do not exist in current kafka cluster

wenbingshen requested a review from chia7712 March 16, 2021 11:00

chia7712 reviewed Mar 17, 2021

View reviewed changes

shenwenbing added 2 commits March 17, 2021 14:00

Merge branch 'trunk' into addErrorLog2LogDirsCommand

3565cca

KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds …

24f520f

…do not exist in current kafka cluster

wenbingshen requested a review from chia7712 March 17, 2021 12:09

dajac changed the title ~~KAFKA-12454:Add ERROR logging on kafka-log-dirs when given brokerIds do not exist in current kafka cluster~~ KAFKA-12454: Add ERROR logging on kafka-log-dirs when given brokerIds do not exist in current kafka cluster Mar 18, 2021

dajac approved these changes Mar 18, 2021

View reviewed changes

chia7712 merged commit bca29e2 into apache:trunk Mar 18, 2021

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

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

Conversation

wenbingshen commented Mar 11, 2021

Committer Checklist (excluded from commit message)

wenbingshen commented Mar 11, 2021

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenbingshen commented Mar 13, 2021

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenbingshen commented Mar 14, 2021

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenbingshen commented Mar 15, 2021

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenbingshen commented Mar 15, 2021

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenbingshen commented Mar 16, 2021

wenbingshen commented Mar 17, 2021

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenbingshen commented Mar 17, 2021

wenbingshen commented Mar 18, 2021

dajac left a comment

Choose a reason for hiding this comment

chia7712 commented Mar 18, 2021