Skip to content

KAFKA-8670: Fix exception for kafka-topics.sh --describe without --topic mentioned #7094

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 7 commits into from
Jul 18, 2019
Merged

KAFKA-8670: Fix exception for kafka-topics.sh --describe without --topic mentioned #7094

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
ccee89d
KAFKA-8670 and KAFKA-8053: Fix kafka-topics.sh --describe without --t…
Jul 15, 2019
225d6fd
Don't change exception message to fix KAFKA-8053, will do that as a s…
Jul 16, 2019
5c156c0
Addressed code review comments to improve variable names, removed red…
Jul 17, 2019
f656b53
Changed name of variable from requireTopicExists to skipIfTopicDoesNo…
Jul 17, 2019
0645516
Renamed variables.
Jul 17, 2019
35993ad
Changed exception message to fix JIRA KAFKA-8053
Jul 17, 2019
ad3c342
Simplified redundant check.
Jul 18, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 15 additions & 12 deletions core/src/main/scala/kafka/admin/TopicCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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()
}

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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, topicOptWithExits)
val requireTopicExists = !(opts.topic.isDefined && opts.ifExists)

@hachikuji hachikuji Jul 18, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Initially I was going to post a comment on the redundant opts.topic.isDefined check here. As I was looking at this, however, I noticed that the second check seemed backwards. I think this has always been broken. We have the following documentation for the --if-exists argument:

if set when altering or deleting or describing topics, the action will only execute if the topic exists.

But we are actually raising the exception only when --if-exists is not defined (i.e. !topicOptWithExists in the original source). I think that is actually the root of the problem. So it seems we just do the following:

ensureTopicExists(topics, opts.topic, opts.ifExists)

Does that seem right?

@wyuka wyuka Jul 18, 2019 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@hachikuji No, I don't think that's right. The --if-exists essentially means "check if the topic exists, and if it does go do this". Which means, we should not be throwing an exception if the topic does not exist. It is like performing a map() operation on an Optional in Java.

If this option is not passed, AND the topic does not exist, we should be throwing an exception.

We can actually simplify line 358 above to

ensureTopicExists(topics, opts.topic, !opts.ifExists)

Let me know your thoughts.

@hachikuji hachikuji Jul 18, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, that makes more sense. So the expected semantic is "do nothing if the topic doesn't exist." Probably the description could be improved. Anyway, the simplification sounds good.

@wyuka wyuka Jul 18, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@hachikuji Thanks. Updated with the simplification. The test servers seem to be having some issue connecting to github.com.

@hachikuji hachikuji Jul 18, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, been seeing that lately. Not sure why.

By the way, I think I get why this logic seemed weird to me. If the topic doesn't exist, we still continue with the operation. But that doesn't make sense, right? Why go on deleting or altering the topic if it doesn't exist? Anyway, this can be improved separately.

ensureTopicExists(topics, opts.topic, requireTopicExists)
val liveBrokers = zkClient.getAllBrokersInCluster.map(_.id).toSet
val describeOptions = new DescribeOptions(opts, liveBrokers)
val adminZkClient = new AdminZkClient(zkClient)
Expand Down Expand Up @@ -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)) {
Expand Down Expand Up @@ -433,14 +433,17 @@ 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 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(topics: Seq[String], topicOptWithExists: Boolean = false) = {
if (topics.isEmpty && !topicOptWithExists) {
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 && requireTopicExists && foundTopics.isEmpty) {
// If given topic doesn't exist then throw exception
throw new IllegalArgumentException(s"Topics in [${topics.mkString(",")}] does not exist")
throw new IllegalArgumentException(s"Topic '${requestedTopic.get}' does not exist as expected")
}
}

Expand Down
5 changes: 5 additions & 0 deletions core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down