[kafka] branch 2.2 updated: KAFKA-8875; CreateTopic API should check topic existence before replication factor (#7298)

[manikumar]

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...@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 <rajinisiva...@googlemail.com>, Jason Gustafson 
<ja...@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")