From ccee89d2f3fa242594af6ed460924e69a5b5418e Mon Sep 17 00:00:00 2001 From: Tirtha Chatterjee Date: Mon, 15 Jul 2019 13:19:21 -0700 Subject: [PATCH 1/7] KAFKA-8670 and KAFKA-8053: Fix kafka-topics.sh --describe without --topic mentioned if there are no topics in cluster, improve error message. If there are no topics in a cluster, kafka-topics.sh --describe without a --topic option should return empty list, do not throw an exception. If there are no matching topics, throw IllegalArgumentException but with better error message. --- .../main/scala/kafka/admin/TopicCommand.scala | 19 +++++++++++-------- .../unit/kafka/admin/TopicCommandTest.scala | 5 +++++ 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 4f529969645e1..7090bc1004031 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -208,7 +208,7 @@ object TopicCommand extends Logging { override def alterTopic(opts: TopicCommandOptions): Unit = { val topic = new CommandTopicPartition(opts) val topics = getTopics(opts.topic, opts.excludeInternalTopics) - ensureTopicExists(topics) + ensureTopicExists(topics, opts.topic) val topicsInfo = adminClient.describeTopics(topics.asJavaCollection).values() adminClient.createPartitions(topics.map {topicName => if (topic.hasReplicaAssignment) { @@ -267,7 +267,7 @@ object TopicCommand extends Logging { override def deleteTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) - ensureTopicExists(topics) + ensureTopicExists(topics, opts.topic) adminClient.deleteTopics(topics.asJavaCollection).all().get() } @@ -317,7 +317,7 @@ object TopicCommand extends Logging { override def alterTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) val tp = new CommandTopicPartition(opts) - ensureTopicExists(topics, opts.ifExists) + ensureTopicExists(topics, opts.topic, opts.ifExists) val adminZkClient = new AdminZkClient(zkClient) topics.foreach { topic => val configs = adminZkClient.fetchEntityConfig(ConfigType.Topic, topic) @@ -355,7 +355,7 @@ object TopicCommand extends Logging { override def describeTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) val topicOptWithExits = opts.topic.isDefined && opts.ifExists - ensureTopicExists(topics, topicOptWithExits) + ensureTopicExists(topics, opts.topic, topicOptWithExits) val liveBrokers = zkClient.getAllBrokersInCluster.map(_.id).toSet val describeOptions = new DescribeOptions(opts, liveBrokers) val adminZkClient = new AdminZkClient(zkClient) @@ -401,7 +401,7 @@ object TopicCommand extends Logging { override def deleteTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) - ensureTopicExists(topics, opts.ifExists) + ensureTopicExists(topics, opts.topic, opts.ifExists) topics.foreach { topic => try { if (Topic.isInternal(topic)) { @@ -437,10 +437,13 @@ object TopicCommand extends Logging { * @param topics * @param topicOptWithExists */ - private def ensureTopicExists(topics: Seq[String], topicOptWithExists: Boolean = false) = { - if (topics.isEmpty && !topicOptWithExists) { + private def ensureTopicExists(topics: Seq[String], desiredTopicName: Option[String], topicOptWithExists: Boolean = false) = { + // If no topic name was mentioned, do not need to throw exception. + if (desiredTopicName.isDefined && topics.isEmpty && !topicOptWithExists) { // If given topic doesn't exist then throw exception - throw new IllegalArgumentException(s"Topics in [${topics.mkString(",")}] does not exist") + if (desiredTopicName.isDefined) { + throw new IllegalArgumentException(s"No topic matching name '${desiredTopicName.get}' found") + } } } diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala index 407c2f3bcc09a..35502f54d0706 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala @@ -416,6 +416,11 @@ class TopicCommandTest extends ZooKeeperTestHarness with Logging with RackAwareT topicService.describeTopic(describeOpts) } + // describe all topics + val describeOptsAllTopics = new TopicCommandOptions(Array()) + // should not throw any error + topicService.describeTopic(describeOptsAllTopics) + // describe topic that does not exist with --if-exists val describeOptsWithExists = new TopicCommandOptions(Array("--topic", testTopicName, "--if-exists")) // should not throw any error From 225d6fd4407e96801598ffd2cef6c02538f134c6 Mon Sep 17 00:00:00 2001 From: Tirtha Chatterjee Date: Mon, 15 Jul 2019 19:04:25 -0700 Subject: [PATCH 2/7] Don't change exception message to fix KAFKA-8053, will do that as a separate commit. --- core/src/main/scala/kafka/admin/TopicCommand.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 7090bc1004031..5c08754060216 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -442,7 +442,7 @@ object TopicCommand extends Logging { if (desiredTopicName.isDefined && topics.isEmpty && !topicOptWithExists) { // If given topic doesn't exist then throw exception if (desiredTopicName.isDefined) { - throw new IllegalArgumentException(s"No topic matching name '${desiredTopicName.get}' found") + throw new IllegalArgumentException(s"Topics in [${topics.mkString(",")}] does not exist") } } } From 5c156c039e5ecea42f00486021ed25de8e1fcfcd Mon Sep 17 00:00:00 2001 From: Tirtha Chatterjee Date: Tue, 16 Jul 2019 23:43:26 -0700 Subject: [PATCH 3/7] Addressed code review comments to improve variable names, removed redundant if check. --- .../main/scala/kafka/admin/TopicCommand.scala | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 5c08754060216..3ce6f78cce906 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -354,8 +354,8 @@ object TopicCommand extends Logging { override def describeTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) - val topicOptWithExits = opts.topic.isDefined && opts.ifExists - ensureTopicExists(topics, opts.topic, topicOptWithExits) + val requireTopicExists = opts.topic.isDefined && opts.ifExists + ensureTopicExists(topics, opts.topic, requireTopicExists) val liveBrokers = zkClient.getAllBrokersInCluster.map(_.id).toSet val describeOptions = new DescribeOptions(opts, liveBrokers) val adminZkClient = new AdminZkClient(zkClient) @@ -433,17 +433,15 @@ object TopicCommand extends Logging { /** * ensures topic existence and throws exception if topic doesn't exist * - * @param opts - * @param topics - * @param topicOptWithExists + * @param foundTopics Topics that were found to match the requested topic name. + * @param requestedTopic Name of the topic that was requested. + * @param requireTopicExists Boolean flag indicating if the topic needs to exist for the operation to be successful. */ - private def ensureTopicExists(topics: Seq[String], desiredTopicName: Option[String], topicOptWithExists: Boolean = false) = { + private def ensureTopicExists(foundTopics: Seq[String], requestedTopic: Option[String], requireTopicExists: Boolean = false) = { // If no topic name was mentioned, do not need to throw exception. - if (desiredTopicName.isDefined && topics.isEmpty && !topicOptWithExists) { + if (requestedTopic.isDefined && foundTopics.isEmpty && !requireTopicExists) { // If given topic doesn't exist then throw exception - if (desiredTopicName.isDefined) { - throw new IllegalArgumentException(s"Topics in [${topics.mkString(",")}] does not exist") - } + throw new IllegalArgumentException(s"Topics in [${foundTopics.mkString(",")}] does not exist") } } From f656b535781df5697351d1b71498bb80eb0f13be Mon Sep 17 00:00:00 2001 From: Tirtha Chatterjee Date: Wed, 17 Jul 2019 00:29:07 -0700 Subject: [PATCH 4/7] Changed name of variable from requireTopicExists to skipIfTopicDoesNotExist --- core/src/main/scala/kafka/admin/TopicCommand.scala | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 3ce6f78cce906..4e6f2b052409d 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -354,8 +354,8 @@ object TopicCommand extends Logging { override def describeTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) - val requireTopicExists = opts.topic.isDefined && opts.ifExists - ensureTopicExists(topics, opts.topic, requireTopicExists) + val skipIfTopicDoesNotExist = opts.topic.isDefined && opts.ifExists + ensureTopicExists(topics, opts.topic, skipIfTopicDoesNotExist) val liveBrokers = zkClient.getAllBrokersInCluster.map(_.id).toSet val describeOptions = new DescribeOptions(opts, liveBrokers) val adminZkClient = new AdminZkClient(zkClient) @@ -435,11 +435,14 @@ object TopicCommand extends Logging { * * @param foundTopics Topics that were found to match the requested topic name. * @param requestedTopic Name of the topic that was requested. - * @param requireTopicExists Boolean flag indicating if the topic needs to exist for the operation to be successful. + * @param skipIfTopicDoesNotExist Indiciates if the command should skip execution + * if a topic with the requested name does not exist. If it is + * set to true, and the requested topic does not exist, the + * command should not throw an exception, but quiety return. */ - private def ensureTopicExists(foundTopics: Seq[String], requestedTopic: Option[String], requireTopicExists: Boolean = false) = { + private def ensureTopicExists(foundTopics: Seq[String], requestedTopic: Option[String], skipIfTopicDoesNotExist: Boolean = false) = { // If no topic name was mentioned, do not need to throw exception. - if (requestedTopic.isDefined && foundTopics.isEmpty && !requireTopicExists) { + if (requestedTopic.isDefined && foundTopics.isEmpty && !skipIfTopicDoesNotExist) { // If given topic doesn't exist then throw exception throw new IllegalArgumentException(s"Topics in [${foundTopics.mkString(",")}] does not exist") } From 06455161f85ef273d6b7270265b78fd63b5a56ad Mon Sep 17 00:00:00 2001 From: Tirtha Chatterjee Date: Wed, 17 Jul 2019 00:35:51 -0700 Subject: [PATCH 5/7] Renamed variables. --- .../main/scala/kafka/admin/TopicCommand.scala | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 4e6f2b052409d..fa9a9e52a6088 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -317,7 +317,7 @@ object TopicCommand extends Logging { override def alterTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) val tp = new CommandTopicPartition(opts) - ensureTopicExists(topics, opts.topic, opts.ifExists) + ensureTopicExists(topics, opts.topic, !opts.ifExists) val adminZkClient = new AdminZkClient(zkClient) topics.foreach { topic => val configs = adminZkClient.fetchEntityConfig(ConfigType.Topic, topic) @@ -354,8 +354,8 @@ object TopicCommand extends Logging { override def describeTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) - val skipIfTopicDoesNotExist = opts.topic.isDefined && opts.ifExists - ensureTopicExists(topics, opts.topic, skipIfTopicDoesNotExist) + val requireTopicExists = !(opts.topic.isDefined && opts.ifExists) + ensureTopicExists(topics, opts.topic, requireTopicExists) val liveBrokers = zkClient.getAllBrokersInCluster.map(_.id).toSet val describeOptions = new DescribeOptions(opts, liveBrokers) val adminZkClient = new AdminZkClient(zkClient) @@ -401,7 +401,7 @@ object TopicCommand extends Logging { override def deleteTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) - ensureTopicExists(topics, opts.topic, opts.ifExists) + ensureTopicExists(topics, opts.topic, !opts.ifExists) topics.foreach { topic => try { if (Topic.isInternal(topic)) { @@ -435,14 +435,13 @@ object TopicCommand extends Logging { * * @param foundTopics Topics that were found to match the requested topic name. * @param requestedTopic Name of the topic that was requested. - * @param skipIfTopicDoesNotExist Indiciates if the command should skip execution - * if a topic with the requested name does not exist. If it is - * set to true, and the requested topic does not exist, the - * command should not throw an exception, but quiety return. + * @param requireTopicExists Indicates if the topic needs to exist for the operation to be successful. + * If set to true, the command will throw an exception if the topic with the + * requested name does not exist. */ - private def ensureTopicExists(foundTopics: Seq[String], requestedTopic: Option[String], skipIfTopicDoesNotExist: Boolean = false) = { + private def ensureTopicExists(foundTopics: Seq[String], requestedTopic: Option[String], requireTopicExists: Boolean = true) = { // If no topic name was mentioned, do not need to throw exception. - if (requestedTopic.isDefined && foundTopics.isEmpty && !skipIfTopicDoesNotExist) { + if (requestedTopic.isDefined && requireTopicExists && foundTopics.isEmpty) { // If given topic doesn't exist then throw exception throw new IllegalArgumentException(s"Topics in [${foundTopics.mkString(",")}] does not exist") } From 35993ada8bc5c5dd564edd6c9979e60c5befb84e Mon Sep 17 00:00:00 2001 From: Tirtha Chatterjee Date: Wed, 17 Jul 2019 13:51:08 -0700 Subject: [PATCH 6/7] Changed exception message to fix JIRA KAFKA-8053 --- core/src/main/scala/kafka/admin/TopicCommand.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index fa9a9e52a6088..bd6f1ad6cbe6c 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -443,7 +443,7 @@ object TopicCommand extends Logging { // If no topic name was mentioned, do not need to throw exception. if (requestedTopic.isDefined && requireTopicExists && foundTopics.isEmpty) { // If given topic doesn't exist then throw exception - throw new IllegalArgumentException(s"Topics in [${foundTopics.mkString(",")}] does not exist") + throw new IllegalArgumentException(s"Topic '${requestedTopic.get}' does not exist as expected") } } From ad3c342ed1ca9e791ec431b63e63ad01e565a71a Mon Sep 17 00:00:00 2001 From: Tirtha Chatterjee Date: Wed, 17 Jul 2019 21:10:53 -0700 Subject: [PATCH 7/7] Simplified redundant check. --- core/src/main/scala/kafka/admin/TopicCommand.scala | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index bd6f1ad6cbe6c..ae3cde87b31c1 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -354,8 +354,7 @@ object TopicCommand extends Logging { override def describeTopic(opts: TopicCommandOptions): Unit = { val topics = getTopics(opts.topic, opts.excludeInternalTopics) - val requireTopicExists = !(opts.topic.isDefined && opts.ifExists) - ensureTopicExists(topics, opts.topic, requireTopicExists) + ensureTopicExists(topics, opts.topic, !opts.ifExists) val liveBrokers = zkClient.getAllBrokersInCluster.map(_.id).toSet val describeOptions = new DescribeOptions(opts, liveBrokers) val adminZkClient = new AdminZkClient(zkClient)