abbccdda commented on a change in pull request #9579:
URL: https://github.com/apache/kafka/pull/9579#discussion_r570568151
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -1084,24 +1087,9 @@ class KafkaApis(val requestChannel: RequestChannel,
(responseTopics ++ unauthorizedResponseStatus).toList
}
- private def createTopic(topic: String,
- numPartitions: Int,
- replicationFactor: Int,
- properties: util.Properties = new
util.Properties()): MetadataResponseTopic = {
- try {
- adminZkClient.createTopic(topic, numPartitions, replicationFactor,
properties, RackAwareMode.Safe)
- info("Auto creation of topic %s with %d partitions and replication
factor %d is successful"
- .format(topic, numPartitions, replicationFactor))
- metadataResponseTopic(Errors.LEADER_NOT_AVAILABLE, topic,
isInternal(topic), util.Collections.emptyList())
- } catch {
- case _: TopicExistsException => // let it go, possibly another broker
created this topic
Review comment:
The problem we have is that `ZkAdminManager.createTopics` only takes a
callback instead of responding to you in realtime whether we hit TopicExists.
Right now we are doing the topic creation async, so unless this is necessary to
be fixed (which today we would just return UNKNOWN_PARTITION which seems to be
semantically similar to LEADER_NOT_AVAILABLE), I think we could just returning
unknown partition immediately without waiting for the async creation?
----------------------------------------------------------------
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:
[email protected]
