[aiven] fix ZK topic creation

aiven · Jan 5, 2024 · 9a60194 · 9a60194
1 parent b2a9e6f
commit 9a60194
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 5 deletions.
diff --git a/core/src/main/scala/kafka/server/ZkAdminManager.scala b/core/src/main/scala/kafka/server/ZkAdminManager.scala
@@ -166,13 +166,14 @@ class ZkAdminManager(val config: KafkaConfig,
       try {
         if (metadataCache.contains(topic.name))
           throw new TopicExistsException(s"Topic '${topic.name}' already exists.")
-        val maybeUuid = Option(topic.id())
-        maybeUuid match {
-          case Some(id) =>
+
+        val maybeUuid = topic.id() match {
+          case Uuid.ZERO_UUID => None
+          case id =>
             if (metadataCache.topicNamesToIds().containsValue(id)) {
               throw new TopicExistsException(s"Topic id '$id' already exists.")
             }
-          case _ =>
+            Some(id)
         }
 
         val nullConfigs = topic.configs.asScala.filter(_.value == null).map(_.name)

diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala
@@ -109,7 +109,7 @@ class AdminZkClient(zkClient: KafkaZkClient,
       validateTopicCreate(topic, partitionReplicaAssignment, config)
 
     info(s"Creating topic $topic with configuration $config and initial partition " +
-      s"assignment $partitionReplicaAssignment")
+      s"assignment $partitionReplicaAssignment, usesTopicId $usesTopicId topicId $maybeTopicId")
 
     // write out the config if there is any, this isn't transactional with the partition assignments
     zkClient.setOrCreateEntityConfigs(ConfigType.Topic, topic, config)