[aiven] feat: add topic id tagged field to create topic api

clients/src/main/resources/common/message/CreateTopicsRequest.json

@@ -53,7 +53,9 @@
           "about": "The configuration name." },
         { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+",
           "about": "The configuration value." }
-      ]}
+      ]},
+      { "name": "Id", "type": "uuid", "tag": 0, "taggedVersions":  "5+", "versions": "5+", "ignorable":  true,
+        "about": "Optional topic id."}
     ]},
     { "name": "timeoutMs", "type": "int32", "versions": "0+", "default": "60000",
       "about": "How long to wait in milliseconds before timing out the request." },

core/src/main/scala/kafka/server/ZkAdminManager.scala

@@ -29,8 +29,7 @@ import org.apache.kafka.admin.AdminUtils
 import org.apache.kafka.clients.admin.{AlterConfigOp, ScramMechanism}
 import org.apache.kafka.common.Uuid
 import org.apache.kafka.common.config.{ConfigDef, ConfigException, ConfigResource}
-import org.apache.kafka.common.errors.ThrottlingQuotaExceededException
-import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidPartitionsException, InvalidReplicaAssignmentException, InvalidRequestException, ReassignmentInProgressException, TopicExistsException, UnknownTopicOrPartitionException, UnsupportedVersionException}
+import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidPartitionsException, InvalidReplicaAssignmentException, InvalidRequestException, ReassignmentInProgressException, ThrottlingQuotaExceededException, TopicExistsException, UnknownTopicOrPartitionException, UnsupportedVersionException}
 import org.apache.kafka.common.message.AlterUserScramCredentialsResponseData.AlterUserScramCredentialsResult
 import org.apache.kafka.common.message.CreatePartitionsRequestData.CreatePartitionsTopic
 import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic
@@ -167,6 +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) =>
+            if (metadataCache.topicNamesToIds().containsValue(id)) {
+              throw new TopicExistsException(s"Topic id '$id' already exists.")
+            }
+          case _ =>
+        }
 
         val nullConfigs = topic.configs.asScala.filter(_.value == null).map(_.name)
         if (nullConfigs.nonEmpty)
@@ -211,7 +218,7 @@ class ZkAdminManager(val config: KafkaConfig,
           CreatePartitionsMetadata(topic.name, assignments.keySet)
         } else {
           controllerMutationQuota.record(assignments.size)
-          adminZkClient.createTopicWithAssignment(topic.name, configs, assignments, validate = false, config.usesTopicId)
+          adminZkClient.createTopicWithAssignment(topic.name, configs, assignments, validate = false, config.usesTopicId, maybeUuid)
           populateIds(includeConfigsAndMetadata, topic.name)
           CreatePartitionsMetadata(topic.name, assignments.keySet)
         }

core/src/main/scala/kafka/zk/AdminZkClient.scala

@@ -102,7 +102,8 @@ class AdminZkClient(zkClient: KafkaZkClient,
                                 config: Properties,
                                 partitionReplicaAssignment: Map[Int, Seq[Int]],
                                 validate: Boolean = true,
-                                usesTopicId: Boolean = false): Unit = {
+                                usesTopicId: Boolean = false,
+                                maybeTopicId: Option[Uuid] = None): Unit = {
     if (validate)
       validateTopicCreate(topic, partitionReplicaAssignment, config)
 
@@ -114,7 +115,7 @@ class AdminZkClient(zkClient: KafkaZkClient,
 
     // create the partition assignment
     writeTopicPartitionAssignment(topic, partitionReplicaAssignment.map { case (k, v) => k -> ReplicaAssignment(v) },
-      isUpdate = false, usesTopicId)
+      isUpdate = false, usesTopicId, maybeTopicId)
   }
 
   /**
@@ -166,12 +167,18 @@ class AdminZkClient(zkClient: KafkaZkClient,
   }
 
   private def writeTopicPartitionAssignment(topic: String, replicaAssignment: Map[Int, ReplicaAssignment],
-                                            isUpdate: Boolean, usesTopicId: Boolean = false): Unit = {
+                                            isUpdate: Boolean, usesTopicId: Boolean = false,
+                                            maybeTopicId: Option[Uuid] = None): Unit = {
     try {
       val assignment = replicaAssignment.map { case (partitionId, replicas) => (new TopicPartition(topic,partitionId), replicas) }.toMap
 
       if (!isUpdate) {
-        val topicIdOpt = if (usesTopicId) Some(Uuid.randomUuid()) else None
+        val topicIdOpt = if (usesTopicId) {
+          maybeTopicId match {
+            case None => Some(Uuid.randomUuid())
+            case _ => maybeTopicId
+          }
+        } else None
         zkClient.createTopicAssignment(topic, topicIdOpt, assignment.map { case (k, v) => k -> v.replicas })
       } else {
         val topicIds = zkClient.getTopicIdsForTopics(Set(topic))

core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala

@@ -26,7 +26,7 @@ import kafka.utils.CoreUtils._
 import kafka.utils.TestUtils._
 import kafka.utils.{Logging, TestUtils}
 import kafka.zk.{AdminZkClient, ConfigEntityTypeZNode, KafkaZkClient}
-import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.{TopicPartition, Uuid}
 import org.apache.kafka.common.config.TopicConfig
 import org.apache.kafka.common.config.internals.QuotaConfigs
 import org.apache.kafka.common.errors.{InvalidReplicaAssignmentException, InvalidTopicException, TopicExistsException}
@@ -53,6 +53,21 @@ class AdminZkClientTest extends QuorumTestHarness with Logging with RackAwareTes
     super.tearDown()
   }
 
+  @Test
+  def testCreateTopicWithId(): Unit = {
+    val brokers = List(0, 1, 2, 3, 4)
+    TestUtils.createBrokersInZk(zkClient, brokers)
+
+    val topicConfig = new Properties()
+
+    val assignment = Map(0 -> List(0, 1, 2),
+      1 -> List(1, 2, 3))
+    val uuid = Uuid.randomUuid
+    adminZkClient.createTopicWithAssignment("test", topicConfig, assignment, usesTopicId = true, maybeTopicId = Some(uuid))
+    val found = zkClient.getTopicIdsForTopics(Set("test"))
+    assertEquals(uuid, found("test"))
+  }
+
   @Test
   def testManualReplicaAssignment(): Unit = {
     val brokers = List(0, 1, 2, 3, 4)

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

@@ -625,6 +625,15 @@ private ApiError createTopic(ControllerRequestContext context,
                                  boolean authorizedToReturnConfigs) {
         Map<String, String> creationConfigs = translateCreationConfigs(topic.configs());
         Map<Integer, PartitionRegistration> newParts = new HashMap<>();
+        Uuid topicId;
+        if (topic.id() != null) {
+            if (topics.containsKey(topic.id())) {
+                return ApiError.fromThrowable(new InvalidTopicException("Topic id " + topic.id() + " already exists"));
+            }
+            topicId = topic.id();
+        } else {
+            topicId = Uuid.randomUuid();
+        }
         if (!topic.assignments().isEmpty()) {
             if (topic.replicationFactor() != -1) {
                 return new ApiError(INVALID_REQUEST,
@@ -725,7 +734,6 @@ private ApiError createTopic(ControllerRequestContext context,
                 numPartitions, e.throttleTimeMs());
             return ApiError.fromThrowable(e);
         }
-        Uuid topicId = Uuid.randomUuid();
         CreatableTopicResult result = new CreatableTopicResult().
             setName(topic.name()).
             setTopicId(topicId).

metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java

@@ -233,12 +233,14 @@ private ReplicationControlTestContext(
             clusterControl.activate();
         }
 
-        CreatableTopicResult createTestTopic(String name,
+        CreatableTopicResult createTestTopic(Uuid id,
+                                             String name,
                                              int numPartitions,
                                              short replicationFactor,
                                              short expectedErrorCode) throws Exception {
             CreateTopicsRequestData request = new CreateTopicsRequestData();
             CreatableTopic topic = new CreatableTopic().setName(name);
+            if (id != null) topic.setId(id);
             topic.setNumPartitions(numPartitions).setReplicationFactor(replicationFactor);
             request.topics().add(topic);
             ControllerRequestContext requestContext = anonymousContextFor(ApiKeys.CREATE_TOPICS);
@@ -253,6 +255,13 @@ CreatableTopicResult createTestTopic(String name,
             return topicResult;
         }
 
+        CreatableTopicResult createTestTopic(String name,
+                                             int numPartitions,
+                                             short replicationFactor,
+                                             short expectedErrorCode) throws Exception {
+            return createTestTopic(null, name, numPartitions, replicationFactor, expectedErrorCode);
+        }
+
         CreatableTopicResult createTestTopic(String name, int[][] replicas) throws Exception {
             return createTestTopic(name, replicas, Collections.emptyMap(), (short) 0);
         }
@@ -752,6 +761,19 @@ public void testInvalidCreateTopicsWithValidateOnlyFlag() throws Exception {
         assertEquals(expectedResponse, result.response());
     }
 
+    @Test
+    public void testCreateTopicsWithId() throws Exception {
+        ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().build();
+        ctx.registerBrokers(0, 1, 2);
+        ctx.unfenceBrokers(0, 1, 2);
+
+        Uuid id = Uuid.randomUuid();
+        CreatableTopicResult initialTopic = ctx.createTestTopic(id, "foo.bar", 2, (short) 2, NONE.code());
+        assertEquals(id, ctx.replicationControl.getTopic(initialTopic.topicId()).topicId());
+        CreatableTopicResult resultWithErrors = ctx.createTestTopic(id, "foo.baz", 2, (short) 2, INVALID_TOPIC_EXCEPTION.code());
+        assertEquals("Topic id " + id + " already exists", resultWithErrors.errorMessage());
+    }
+
     @Test
     public void testCreateTopicsWithPolicy() throws Exception {
         MockCreateTopicPolicy createTopicPolicy = new MockCreateTopicPolicy(asList(