kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From maniku...@apache.org
Subject [kafka] branch 2.2 updated: KAFKA-8875; CreateTopic API should check topic existence before replication factor (#7298)
Date Wed, 13 Nov 2019 10:40:19 GMT
This is an automated email from the ASF dual-hosted git repository.

manikumar pushed a commit to branch 2.2
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.2 by this push:
     new b9adf4c  KAFKA-8875; CreateTopic API should check topic existence before replication
factor (#7298)
b9adf4c is described below

commit b9adf4c0a9d4b4a510a281278679d0ee68b97566
Author: huxi <huxi_2b@hotmail.com>
AuthorDate: Thu Sep 12 05:24:27 2019 +0800

    KAFKA-8875; CreateTopic API should check topic existence before replication factor (#7298)
    
    If the topic already exists, `handleCreateTopicsRequest` should return TopicExistsException
even given an invalid config (replication factor for instance).
    
    Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>, Jason Gustafson <jason@confluent.io>
---
 core/src/main/scala/kafka/server/AdminManager.scala     |  5 ++++-
 .../kafka/api/AdminClientIntegrationTest.scala          | 17 +++++++++++++++++
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 4850eb4..d526329 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -25,7 +25,7 @@ import kafka.metrics.KafkaMetricsGroup
 import kafka.utils._
 import kafka.zk.{AdminZkClient, KafkaZkClient}
 import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource}
-import org.apache.kafka.common.errors.{ApiException, InvalidPartitionsException, InvalidReplicaAssignmentException,
InvalidRequestException, ReassignmentInProgressException, UnknownTopicOrPartitionException,
InvalidConfigurationException}
+import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidPartitionsException,
InvalidReplicaAssignmentException, InvalidRequestException, ReassignmentInProgressException,
TopicExistsException, UnknownTopicOrPartitionException}
 import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.ListenerName
@@ -80,6 +80,9 @@ class AdminManager(val config: KafkaConfig,
     val brokers = metadataCache.getAliveBrokers.map { b => kafka.admin.BrokerMetadata(b.id,
b.rack) }
     val metadata = createInfo.map { case (topic, arguments) =>
       try {
+        if (metadataCache.contains(topic))
+          throw new TopicExistsException(s"Topic '$topic' already exists.")
+
         val configs = new Properties()
         arguments.configs.asScala.foreach { case (key, value) =>
           configs.setProperty(key, value)
diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index 8f8a679..19e6b97 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -211,6 +211,23 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with
Logging {
   }
 
   @Test
+  def testCreateExistingTopicsThrowTopicExistsException(): Unit = {
+    client = AdminClient.create(createConfig())
+    val topic = "mytopic"
+    val topics = Seq(topic)
+    val newTopics = Seq(new NewTopic(topic, 1, 1.toShort))
+
+    client.createTopics(newTopics.asJava).all.get()
+    waitForTopics(client, topics, List())
+
+    val newTopicsWithInvalidRF = Seq(new NewTopic(topic, 1, (servers.size + 1).toShort))
+    val e = intercept[ExecutionException] {
+      client.createTopics(newTopicsWithInvalidRF.asJava, new CreateTopicsOptions().validateOnly(true)).all.get()
+    }
+    assertTrue(e.getCause.isInstanceOf[TopicExistsException])
+  }
+
+  @Test
   def testMetadataRefresh(): Unit = {
     client = AdminClient.create(createConfig())
     val topics = Seq("mytopic")


Mime
View raw message