Skip to content

KAFKA-9945: TopicCommand should support --if-exists and --if-not-exists when --bootstrap-server is used #8737

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
cmccabe merged 3 commits into from
Jun 2, 2020
Merged

KAFKA-9945: TopicCommand should support --if-exists and --if-not-exists when --bootstrap-server is used #8737

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c18de27
KAFKA-9945: TopicCommand should support --if-exists and --if-not-exis…
May 28, 2020
273c846
KAFKA-9945: Address CR comments
May 29, 2020
490085d
KAFKA-9945: More CR feedback
May 30, 2020
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
112 changes: 64 additions & 48 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 @@ -228,7 +228,7 @@ object TopicCommand extends Logging {
if (topic.partitions.exists(partitions => partitions < 1))
throw new IllegalArgumentException(s"The partitions must be greater than 0")

if (!adminClient.listTopics().names().get().contains(topic.name)) {
try {
Comment thread
cmccabe marked this conversation as resolved.
val newTopic = if (topic.hasReplicaAssignment)
new NewTopic(topic.name, asJavaReplicaReassignment(topic.replicaAssignment.get))
else {
Expand All @@ -247,8 +247,12 @@ object TopicCommand extends Logging {
val createResult = adminClient.createTopics(Collections.singleton(newTopic))
createResult.all().get()
println(s"Created topic ${topic.name}.")
} else {
throw new IllegalArgumentException(s"Topic ${topic.name} already exists")
} catch {
case e : ExecutionException =>
if (e.getCause == null)
throw e
if (!(e.getCause.isInstanceOf[TopicExistsException] && topic.ifTopicDoesntExist()))
throw e.getCause
}
}

Expand All @@ -259,24 +263,29 @@ object TopicCommand extends Logging {
override def alterTopic(opts: TopicCommandOptions): Unit = {
val topic = new CommandTopicPartition(opts)
val topics = getTopics(opts.topic, opts.excludeInternalTopics)
ensureTopicExists(topics, opts.topic)
val topicsInfo = adminClient.describeTopics(topics.asJavaCollection).values()
adminClient.createPartitions(topics.map {topicName =>
if (topic.hasReplicaAssignment) {
val startPartitionId = topicsInfo.get(topicName).get().partitions().size()
val newAssignment = {
val replicaMap = topic.replicaAssignment.get.drop(startPartitionId)
new util.ArrayList(replicaMap.map(p => p._2.asJava).asJavaCollection).asInstanceOf[util.List[util.List[Integer]]]
ensureTopicExists(topics, opts.topic, !opts.ifExists)
Comment thread
cmccabe marked this conversation as resolved.

if (topics.nonEmpty) {
val topicsInfo = adminClient.describeTopics(topics.asJavaCollection).values()
adminClient.createPartitions(topics.map { topicName =>
if (topic.hasReplicaAssignment) {
val startPartitionId = topicsInfo.get(topicName).get().partitions().size()
val newAssignment = {
val replicaMap = topic.replicaAssignment.get.drop(startPartitionId)
new util.ArrayList(replicaMap.map(p => p._2.asJava).asJavaCollection).asInstanceOf[util.List[util.List[Integer]]]
}
topicName -> NewPartitions.increaseTo(topic.partitions.get, newAssignment)
} else {
topicName -> NewPartitions.increaseTo(topic.partitions.get)
}
topicName -> NewPartitions.increaseTo(topic.partitions.get, newAssignment)
} else {
topicName -> NewPartitions.increaseTo(topic.partitions.get)
}}.toMap.asJava).all().get()
}.toMap.asJava).all().get()
}
}

private def listAllReassignments(): Map[TopicPartition, PartitionReassignment] = {
private def listAllReassignments(topicPartitions: util.Set[TopicPartition]): Map[TopicPartition, PartitionReassignment] = {
try {
adminClient.listPartitionReassignments(new ListPartitionReassignmentsOptions).reassignments().get().asScala
adminClient.listPartitionReassignments(topicPartitions, new ListPartitionReassignmentsOptions)
.reassignments().get().asScala
} catch {
case e: ExecutionException =>
e.getCause match {
Expand All @@ -290,41 +299,48 @@ object TopicCommand extends Logging {

override def describeTopic(opts: TopicCommandOptions): Unit = {
val topics = getTopics(opts.topic, opts.excludeInternalTopics)
val allConfigs = adminClient.describeConfigs(topics.map(new ConfigResource(Type.TOPIC, _)).asJavaCollection).values()
val liveBrokers = adminClient.describeCluster().nodes().get().asScala.map(_.id())
val reassignments = listAllReassignments()
val topicDescriptions = adminClient.describeTopics(topics.asJavaCollection).all().get().values().asScala
val describeOptions = new DescribeOptions(opts, liveBrokers.toSet)

for (td <- topicDescriptions) {
val topicName = td.name
val config = allConfigs.get(new ConfigResource(Type.TOPIC, topicName)).get()
val sortedPartitions = td.partitions.asScala.sortBy(_.partition)

if (describeOptions.describeConfigs) {
val hasNonDefault = config.entries().asScala.exists(!_.isDefault)
if (!opts.reportOverriddenConfigs || hasNonDefault) {
val numPartitions = td.partitions().size
val firstPartition = td.partitions.iterator.next()
val reassignment = reassignments.get(new TopicPartition(td.name, firstPartition.partition))
val topicDesc = TopicDescription(topicName, numPartitions, getReplicationFactor(firstPartition, reassignment), config, markedForDeletion = false)
topicDesc.printDescription()
ensureTopicExists(topics, opts.topic, !opts.ifExists)
Comment thread
vinothchandar marked this conversation as resolved.

if (topics.nonEmpty) {

@cmccabe cmccabe May 30, 2020

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.

It's not necessary to check this. AdminClient handles being passed empty lists or sets of topics.

@vinothchandar vinothchandar May 30, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

val liveBrokers = adminClient.describeCluster().nodes().get().asScala.map(_.id())

This line still makes an RPC call.

val allConfigs = adminClient.describeConfigs(topics.map(new ConfigResource(Type.TOPIC, _)).asJavaCollection).values()
val liveBrokers = adminClient.describeCluster().nodes().get().asScala.map(_.id())
val topicDescriptions = adminClient.describeTopics(topics.asJavaCollection).all().get().values().asScala
val describeOptions = new DescribeOptions(opts, liveBrokers.toSet)
val topicPartitions = topicDescriptions
.flatMap(td => td.partitions.iterator().asScala.map(p => new TopicPartition(td.name(), p.partition())))
.toSet.asJava
val reassignments = listAllReassignments(topicPartitions)

for (td <- topicDescriptions) {
val topicName = td.name
val config = allConfigs.get(new ConfigResource(Type.TOPIC, topicName)).get()
val sortedPartitions = td.partitions.asScala.sortBy(_.partition)

if (describeOptions.describeConfigs) {
val hasNonDefault = config.entries().asScala.exists(!_.isDefault)
if (!opts.reportOverriddenConfigs || hasNonDefault) {
val numPartitions = td.partitions().size
val firstPartition = td.partitions.iterator.next()
val reassignment = reassignments.get(new TopicPartition(td.name, firstPartition.partition))
val topicDesc = TopicDescription(topicName, numPartitions, getReplicationFactor(firstPartition, reassignment), config, markedForDeletion = false)
topicDesc.printDescription()
}
}
}

if (describeOptions.describePartitions) {
for (partition <- sortedPartitions) {
val reassignment = reassignments.get(new TopicPartition(td.name, partition.partition))
val partitionDesc = PartitionDescription(topicName, partition, Some(config), markedForDeletion = false, reassignment)
describeOptions.maybePrintPartitionDescription(partitionDesc)
if (describeOptions.describePartitions) {
for (partition <- sortedPartitions) {
val reassignment = reassignments.get(new TopicPartition(td.name, partition.partition))
val partitionDesc = PartitionDescription(topicName, partition, Some(config), markedForDeletion = false, reassignment)
describeOptions.maybePrintPartitionDescription(partitionDesc)
}
}
}
}
}

override def deleteTopic(opts: TopicCommandOptions): Unit = {
val topics = getTopics(opts.topic, opts.excludeInternalTopics)
ensureTopicExists(topics, opts.topic)
ensureTopicExists(topics, opts.topic, !opts.ifExists)
Comment thread
vinothchandar marked this conversation as resolved.
adminClient.deleteTopics(topics.asJavaCollection).all().get()
}

Expand Down Expand Up @@ -501,7 +517,7 @@ object TopicCommand extends Logging {
* 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], requireTopicExists: Boolean = true): Unit = {
private def ensureTopicExists(foundTopics: Seq[String], requestedTopic: Option[String], requireTopicExists: Boolean): Unit = {
// 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
Expand Down Expand Up @@ -639,9 +655,9 @@ object TopicCommand extends Logging {
private val topicsWithOverridesOpt = parser.accepts("topics-with-overrides",
"if set when describing topics, only show topics that have overridden configs")
private val ifExistsOpt = parser.accepts("if-exists",
"if set when altering or deleting or describing topics, the action will only execute if the topic exists. Not supported with the --bootstrap-server option.")
"if set when altering or deleting or describing topics, the action will only execute if the topic exists.")
private val ifNotExistsOpt = parser.accepts("if-not-exists",
"if set when creating topics, the action will only execute if the topic does not already exist. Not supported with the --bootstrap-server option.")
"if set when creating topics, the action will only execute if the topic does not already exist.")

private val disableRackAware = parser.accepts("disable-rack-aware", "Disable rack aware replica assignment")

Expand Down Expand Up @@ -736,8 +752,8 @@ object TopicCommand extends Logging {
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts - reportUnavailablePartitionsOpt + topicsWithOverridesOpt)
CommandLineUtils.checkInvalidArgs(parser, options, topicsWithOverridesOpt,
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts)
CommandLineUtils.checkInvalidArgs(parser, options, ifExistsOpt, allTopicLevelOpts -- Set(alterOpt, deleteOpt, describeOpt) ++ Set(bootstrapServerOpt))
CommandLineUtils.checkInvalidArgs(parser, options, ifNotExistsOpt, allTopicLevelOpts -- Set(createOpt) ++ Set(bootstrapServerOpt))
Comment thread
vinothchandar marked this conversation as resolved.
CommandLineUtils.checkInvalidArgs(parser, options, ifExistsOpt, allTopicLevelOpts -- Set(alterOpt, deleteOpt, describeOpt))
CommandLineUtils.checkInvalidArgs(parser, options, ifNotExistsOpt, allTopicLevelOpts -- Set(createOpt))
CommandLineUtils.checkInvalidArgs(parser, options, excludeInternalTopicOpt, allTopicLevelOpts -- Set(listOpt, describeOpt))
}
}
Expand Down
38 changes: 31 additions & 7 deletions core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin._
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.config.{ConfigException, ConfigResource, TopicConfig}
import org.apache.kafka.common.errors.TopicExistsException
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.Errors
Expand Down Expand Up @@ -214,7 +215,7 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
}

@Test
def testCreateIfItAlreadyExists(): Unit = {
def testCreateWhenAlreadyExists(): Unit = {
val numPartitions = 1

// create the topic
Expand All @@ -223,11 +224,18 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
createAndWaitTopic(createOpts)

// try to re-create the topic
intercept[IllegalArgumentException] {
intercept[TopicExistsException] {
topicService.createTopic(createOpts)
}
}

@Test
def testCreateWhenAlreadyExistsWithIfNotExists(): Unit = {
val createOpts = new TopicCommandOptions(Array("--topic", testTopicName, "--if-not-exists"))
createAndWaitTopic(createOpts)
topicService.createTopic(createOpts)
}

@Test
def testCreateWithReplicaAssignment(): Unit = {
// create the topic
Expand Down Expand Up @@ -431,10 +439,9 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
}

@Test
def testIfExistsAndIfNotExistsOptionsInvalidWithBootstrapServers(): Unit = {
// alter a topic that does not exist without --if-exists
assertCheckArgsExitCode(1, new TopicCommandOptions(Array("--bootstrap-server", "server1:9092", "--alter", "--if-exists", "--topic", testTopicName, "--partitions", "1")))
assertCheckArgsExitCode(1, new TopicCommandOptions(Array("--bootstrap-server", "server1:9092", "--create", "--if-not-exists", "--topic", testTopicName, "--partitions", "1", "--replication-factor", "1")))
def testAlterWhenTopicDoesntExistWithIfExists(): Unit = {
topicService.alterTopic(new TopicCommandOptions(
Array("--topic", testTopicName, "--partitions", "1", "--if-exists")))
}

@Test
Expand Down Expand Up @@ -530,14 +537,19 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
}

@Test
def testDeleteIfExists(): Unit = {
def testDeleteWhenTopicDoesntExist(): Unit = {
// delete a topic that does not exist
val deleteOpts = new TopicCommandOptions(Array("--topic", testTopicName))
intercept[IllegalArgumentException] {
topicService.deleteTopic(deleteOpts)
}
}

@Test
def testDeleteWhenTopicDoesntExistWithIfExists(): Unit = {
topicService.deleteTopic(new TopicCommandOptions(Array("--topic", testTopicName, "--if-exists")))
}

@Test
def testDescribe(): Unit = {
adminClient.createTopics(
Expand All @@ -551,6 +563,18 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
rows(0).startsWith(s"Topic:$testTopicName\tPartitionCount:2")
}

@Test
def testDescribeWhenTopicDoesntExist(): Unit = {
intercept[IllegalArgumentException] {
topicService.describeTopic(new TopicCommandOptions(Array("--topic", testTopicName)))
}
}

@Test
def testDescribeWhenTopicDoesntExistWithIfExists(): Unit = {
topicService.describeTopic(new TopicCommandOptions(Array("--topic", testTopicName, "--if-exists")))
}

@Test
def testDescribeUnavailablePartitions(): Unit = {
adminClient.createTopics(
Expand Down