diff --git a/clients/src/main/resources/common/message/CreateTopicsRequest.json b/clients/src/main/resources/common/message/CreateTopicsRequest.json index 0882de9fa64d..01f38497428d 100644 --- a/clients/src/main/resources/common/message/CreateTopicsRequest.json +++ b/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." }, diff --git a/core/src/main/scala/kafka/server/ZkAdminManager.scala b/core/src/main/scala/kafka/server/ZkAdminManager.scala index ac7a82dc0dfa..47186b61903f 100644 --- a/core/src/main/scala/kafka/server/ZkAdminManager.scala +++ b/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) } diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala index 3a4715660fbc..94d86656cc24 100644 --- a/core/src/main/scala/kafka/zk/AdminZkClient.scala +++ b/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)) diff --git a/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala b/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala index e3171647ff18..696691d6caa4 100644 --- a/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala +++ b/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) diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index 8e6e4a20d220..5910de2903a1 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -625,6 +625,15 @@ private ApiError createTopic(ControllerRequestContext context, boolean authorizedToReturnConfigs) { Map creationConfigs = translateCreationConfigs(topic.configs()); Map 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). diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index 41bffdba92fc..4ac9759d0ea3 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/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(