chia7712 commented on a change in pull request #10465: URL: https://github.com/apache/kafka/pull/10465#discussion_r607623340
########## File path: core/src/test/java/kafka/test/MockController.java ########## @@ -77,8 +82,28 @@ private MockController(Collection<MockTopic> initialTopics) { } @Override - public CompletableFuture<CreateTopicsResponseData> createTopics(CreateTopicsRequestData request) { - throw new UnsupportedOperationException(); + synchronized public CompletableFuture<CreateTopicsResponseData> + createTopics(CreateTopicsRequestData request) { + CreateTopicsResponseData response = new CreateTopicsResponseData(); + for (CreatableTopic topic : request.topics()) { + if (topics.containsKey(topic.name())) { + response.topics().add(new CreatableTopicResult(). + setName(topic.name()). + setErrorCode(Errors.TOPIC_ALREADY_EXISTS.code())); + } else { + long topicId = nextTopicId.getAndAdd(1); + Uuid topicUuid = new Uuid(0, topicId); + topicNameToId.put(topic.name(), topicUuid); + topics.put(topicUuid, new MockTopic(topic.name(), topicUuid)); + response.topics().add(new CreatableTopicResult(). + setName(topic.name()). + setErrorCode(Errors.NONE.code()). + setTopicId(topicUuid)); + // For a better mock, we might want to return configs, replication Review comment: agreed. why not including such information? ########## File path: core/src/test/java/kafka/test/MockController.java ########## @@ -77,8 +82,28 @@ private MockController(Collection<MockTopic> initialTopics) { } @Override - public CompletableFuture<CreateTopicsResponseData> createTopics(CreateTopicsRequestData request) { - throw new UnsupportedOperationException(); + synchronized public CompletableFuture<CreateTopicsResponseData> + createTopics(CreateTopicsRequestData request) { + CreateTopicsResponseData response = new CreateTopicsResponseData(); + for (CreatableTopic topic : request.topics()) { + if (topics.containsKey(topic.name())) { Review comment: the key type of `topics` is `Uuid` and hence this check is weird. Maybe it should be replaced by `topicNameToId` ########## File path: core/src/test/scala/unit/kafka/server/ControllerApisTest.scala ########## @@ -32,10 +32,13 @@ import org.apache.kafka.common.Uuid.ZERO_UUID import org.apache.kafka.common.errors.{InvalidRequestException, NotControllerException, TopicDeletionDisabledException} import org.apache.kafka.common.memory.MemoryPool import org.apache.kafka.common.message.ApiMessageType.ListenerType +import org.apache.kafka.common.message.CreateTopicsRequestData.{CreatableTopic, CreatableTopicCollection} +import org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult import org.apache.kafka.common.message.DeleteTopicsRequestData.DeleteTopicState import org.apache.kafka.common.message.DeleteTopicsResponseData.DeletableTopicResult -import org.apache.kafka.common.message.{BrokerRegistrationRequestData, DeleteTopicsRequestData} +import org.apache.kafka.common.message.{BrokerRegistrationRequestData, CreateTopicsRequestData, DeleteTopicsRequestData} import org.apache.kafka.common.network.{ClientInformation, ListenerName} +import org.apache.kafka.common.protocol.Errors.{INVALID_REQUEST, NONE, TOPIC_AUTHORIZATION_FAILED} Review comment: The code `Errors.TOPIC_AUTHORIZATION_FAILED` still exists in this file. Could you remove that redundant prefix? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org