Fix a few lingering problems

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

@@ -17,27 +17,29 @@
 
 package kafka.server
 
-import java.util
-import java.util.Properties
+import java.util.{Collections, Properties}
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicReference
 
 import kafka.cluster.Broker
 import kafka.controller.KafkaController
 import kafka.coordinator.group.GroupCoordinator
 import kafka.coordinator.transaction.TransactionCoordinator
 import kafka.utils.Logging
 import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.common.errors.InvalidTopicException
 import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.internals.Topic.{GROUP_METADATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME}
 import org.apache.kafka.common.message.CreateTopicsRequestData
 import org.apache.kafka.common.message.CreateTopicsRequestData.{CreatableTopic, CreateableTopicConfig, CreateableTopicConfigCollection}
-import org.apache.kafka.common.message.MetadataResponseData.{MetadataResponsePartition, MetadataResponseTopic}
+import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.CreateTopicsRequest
+import org.apache.kafka.common.requests.{ApiError, CreateTopicsRequest}
 import org.apache.kafka.common.utils.Time
 
-import scala.collection.{Map, Seq, Set}
+import scala.collection.{Map, Seq, Set, mutable}
+import scala.jdk.CollectionConverters._
 
 trait AutoTopicCreationManager {
 
@@ -80,12 +82,12 @@ object AutoTopicCreationManager {
         ))
       else
         None
-    new AutoTopicCreationManagerImpl(config, metadataCache, channelManager, adminManager,
+    new DefaultAutoTopicCreationManager(config, metadataCache, channelManager, adminManager,
       controller, groupCoordinator, txnCoordinator)
   }
 }
 
-class AutoTopicCreationManagerImpl(
+class DefaultAutoTopicCreationManager(
   config: KafkaConfig,
   metadataCache: MetadataCache,
   channelManager: Option[BrokerToControllerChannelManager],
@@ -95,7 +97,7 @@ class AutoTopicCreationManagerImpl(
   txnCoordinator: TransactionCoordinator
 ) extends AutoTopicCreationManager with Logging {
 
-  private val inflightTopics = new ConcurrentHashMap[String, CreatableTopic]()
+  private val inflightTopics = Collections.newSetFromMap(new ConcurrentHashMap[String, java.lang.Boolean]())
 
   override def start(): Unit = {
     channelManager.foreach(_.start())
@@ -106,71 +108,111 @@ class AutoTopicCreationManagerImpl(
     inflightTopics.clear()
   }
 
-  override def createTopics(topics: Set[String],
-                            controllerMutationQuota: ControllerMutationQuota): Seq[MetadataResponseTopic] = {
-    val topicResponses = topics.map(topic =>
-      metadataResponseTopic(
-      if (!hasEnoughAliveBrokers(topic))
-        Errors.INVALID_REPLICATION_FACTOR
-      else
-        Errors.UNKNOWN_TOPIC_OR_PARTITION,
-      topic, Topic.isInternal(topic), util.Collections.emptyList()))
-
-    val topicConfigs = topicResponses
-      .filter(topicResponse => shouldCreate(topicResponse))
-      .map(topicResponse => getTopicConfigs(topicResponse.name()))
-      .map(topic => (topic.name, topic)).toMap
-
-    if (topicConfigs.nonEmpty) {
-      if (!controller.isActive && channelManager.isDefined) {
-        // Mark the topics as inflight during auto creation through forwarding.
-        topicConfigs.foreach(config => inflightTopics.put(config._1, config._2))
-
-        val topicsToCreate = new CreateTopicsRequestData.CreatableTopicCollection
-        topicConfigs.foreach(config => topicsToCreate.add(config._2))
-        val createTopicsRequest = new CreateTopicsRequest.Builder(
-          new CreateTopicsRequestData()
-            .setTimeoutMs(config.requestTimeoutMs)
-            .setTopics(topicsToCreate)
-        )
-
-        channelManager.get.sendRequest(createTopicsRequest, new ControllerRequestCompletionHandler {
-          override def onTimeout(): Unit = {
-            debug(s"Auto topic creation timed out for $topics.")
-            clearInflightRequests(topicConfigs)
-          }
+  override def createTopics(
+    topics: Set[String],
+    controllerMutationQuota: ControllerMutationQuota
+  ): Seq[MetadataResponseTopic] = {
+    val (creatableTopics, uncreatableTopicResponses) = filterCreatableTopics(topics)
 
-          override def onComplete(response: ClientResponse): Unit = {
-            debug(s"Auto topic creation completed for $topics.")
-            clearInflightRequests(topicConfigs)
+    val creatableTopicResponses = if (creatableTopics.isEmpty) {
+      Seq.empty
+    } else if (!controller.isActive && channelManager.isDefined) {
+      sendCreateTopicRequest(creatableTopics)
+    } else {
+      createTopicsInZk(creatableTopics, controllerMutationQuota)
+    }
+
+    uncreatableTopicResponses ++ creatableTopicResponses
+  }
+
+  private def createTopicsInZk(
+    creatableTopics: Map[String, CreatableTopic],
+    controllerMutationQuota: ControllerMutationQuota
+  ): Seq[MetadataResponseTopic] = {
+    val topicErrors = new AtomicReference[Map[String, ApiError]]()
+
+    // Note that we use timeout = 0 since we do not need to wait for metadata propagation
+    // and we want to get the response error immediately.
+    adminManager.createTopics(
+      timeout = 0,
+      validateOnly = false,
+      creatableTopics,
+      Map.empty,
+      controllerMutationQuota,
+      topicErrors.set
+    )
+
+    val creatableTopicResponses = Option(topicErrors.get) match {
+      case Some(errors) =>
+        errors.toSeq.map { case (topic, apiError) =>
+          val error = apiError.error match {
+            case Errors.TOPIC_ALREADY_EXISTS | Errors.REQUEST_TIMED_OUT =>
+              // The timeout error is expected because we set timeout=0. This
+              // nevertheless indicates that the topic metadata was created
+              // successfully, so we return LEADER_NOT_AVAILABLE.
+              Errors.LEADER_NOT_AVAILABLE
+            case error => error
           }
-        })
-        info(s"Sent $topics to the active controller for auto creation.")
-      } else {
-        adminManager.createTopics(
-          config.requestTimeoutMs,
-          validateOnly = false,
-          topicConfigs,
-          Map.empty,
-          controllerMutationQuota,
-          _ => ())
-        info(s"Topics $topics are being created asynchronously.")
-      }
+
+          new MetadataResponseTopic()
+            .setErrorCode(error.code)
+            .setName(topic)
+            .setIsInternal(Topic.isInternal(topic))
+        }
+
+      case None =>
+        creatableTopics.keySet.toSeq.map { topic =>
+          new MetadataResponseTopic()
+            .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
+            .setName(topic)
+            .setIsInternal(Topic.isInternal(topic))
+        }
     }
-    topicResponses.toSeq
+
+    creatableTopicResponses
   }
 
-  private def shouldCreate(topicResponse: MetadataResponseTopic): Boolean = {
-    Errors.forCode(topicResponse.errorCode()) == Errors.UNKNOWN_TOPIC_OR_PARTITION &&
-      !inflightTopics.containsKey(topicResponse.name)
+  private def sendCreateTopicRequest(
+    creatableTopics: Map[String, CreatableTopic]
+  ): Seq[MetadataResponseTopic] = {
+    val topicsToCreate = new CreateTopicsRequestData.CreatableTopicCollection(creatableTopics.size)
+    topicsToCreate.addAll(creatableTopics.values.asJavaCollection)
+
+    val createTopicsRequest = new CreateTopicsRequest.Builder(
+      new CreateTopicsRequestData()
+        .setTimeoutMs(config.requestTimeoutMs)
+        .setTopics(topicsToCreate)
+    )
+
+    channelManager.get.sendRequest(createTopicsRequest, new ControllerRequestCompletionHandler {
+      override def onTimeout(): Unit = {
+        debug(s"Auto topic creation timed out for ${creatableTopics.keys}.")
+        clearInflightRequests(creatableTopics)
+      }
+
+      override def onComplete(response: ClientResponse): Unit = {
+        debug(s"Auto topic creation completed for ${creatableTopics.keys}.")
+        clearInflightRequests(creatableTopics)
+      }
+    })
+
+    val creatableTopicResponses = creatableTopics.keySet.toSeq.map { topic =>
+      new MetadataResponseTopic()
+        .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
+        .setName(topic)
+        .setIsInternal(Topic.isInternal(topic))
+    }
+
+    info(s"Sent auto-creation request for ${creatableTopics.keys} to the active controller.")
+    creatableTopicResponses
   }
 
-  private def clearInflightRequests(topicConfigs: Map[String, CreatableTopic]): Unit = {
-    topicConfigs.foreach(config => inflightTopics.remove(config._1))
-    debug(s"Cleared pending topic creation states for $topicConfigs")
+  private def clearInflightRequests(creatableTopics: Map[String, CreatableTopic]): Unit = {
+    creatableTopics.keySet.foreach(inflightTopics.remove)
+    debug(s"Cleared inflight topic creation state for $creatableTopics")
   }
 
-  private def getTopicConfigs(topic: String): CreatableTopic = {
+  private def creatableTopic(topic: String): CreatableTopic = {
     topic match {
       case GROUP_METADATA_TOPIC_NAME =>
         new CreatableTopic()
@@ -204,48 +246,74 @@ class AutoTopicCreationManagerImpl(
     topicConfigs
   }
 
-  private def hasEnoughAliveBrokers(topic: String): Boolean = {
-    if (topic == null)
-      throw new IllegalArgumentException("topic must not be null")
+  private def isValidTopicName(topic: String): Boolean = {
+    try {
+      Topic.validate(topic)
+      true
+    } catch {
+      case e: InvalidTopicException =>
+        false
+    }
+  }
+
+  private def filterCreatableTopics(
+    topics: Set[String]
+  ): (Map[String, CreatableTopic], Seq[MetadataResponseTopic]) = {
 
     val aliveBrokers = metadataCache.getAliveBrokers
+    val creatableTopics = mutable.Map.empty[String, CreatableTopic]
+    val uncreatableTopics = mutable.Buffer.empty[MetadataResponseTopic]
 
-    topic match {
+    topics.foreach { topic =>
+      // Attempt basic topic validation before sending any requests to the controller.
+      val validationError: Option[Errors] = if (!isValidTopicName(topic)) {
+        Some(Errors.INVALID_TOPIC_EXCEPTION)
+      } else if (!hasEnoughLiveBrokers(topic, aliveBrokers)) {
+        Some(Errors.INVALID_REPLICATION_FACTOR)
+      } else if (!inflightTopics.add(topic)) {
+        Some(Errors.UNKNOWN_TOPIC_OR_PARTITION)
+      } else {
+        None
+      }
+
+      validationError match {
+        case Some(error) =>
+          uncreatableTopics.addOne(new MetadataResponseTopic()
+            .setErrorCode(error.code)
+            .setName(topic)
+            .setIsInternal(Topic.isInternal(topic))
+          )
+        case None =>
+          creatableTopics.put(topic, creatableTopic(topic))
+      }
+    }
+
+    (creatableTopics, uncreatableTopics)
+  }
+
+  private def hasEnoughLiveBrokers(
+    topicName: String,
+    aliveBrokers: Seq[Broker]
+  ): Boolean = {
+    val (replicationFactor, replicationFactorConfig) = topicName match {
       case GROUP_METADATA_TOPIC_NAME =>
-        checkEnoughLiveBrokers(aliveBrokers, config.offsetsTopicReplicationFactor.intValue(), topic, None)
+        (config.offsetsTopicReplicationFactor.intValue, KafkaConfig.OffsetsTopicReplicationFactorProp)
+
       case TRANSACTION_STATE_TOPIC_NAME =>
-        checkEnoughLiveBrokers(aliveBrokers, config.transactionTopicReplicationFactor.intValue(), topic, None)
+        (config.transactionTopicReplicationFactor.intValue, KafkaConfig.TransactionsTopicReplicationFactorProp)
+
       case _ =>
-        checkEnoughLiveBrokers(aliveBrokers, config.defaultReplicationFactor, topic, None)
+        (config.defaultReplicationFactor, KafkaConfig.DefaultReplicationFactorProp)
     }
-  }
 
-  private def checkEnoughLiveBrokers(aliveBrokers: Seq[Broker],
-                                     replicationFactor: Int,
-                                     topicName: String,
-                                     configName: Option[String]): Boolean = {
     if (aliveBrokers.size < replicationFactor) {
-      val configHint = configName match {
-        case Some(config) => s", whose replication factor is configured via $config"
-        case None => ""
-      }
       error(s"Number of alive brokers '${aliveBrokers.size}' does not meet the required replication factor " +
-        s"'$replicationFactor' for auto creation of topic '$topicName'$configHint. " +
-        s"This error can be ignored if the cluster is starting up " +
-        s"and not all brokers are up yet.")
+        s"'$replicationFactor' for auto creation of topic '$topicName' which is configured by $replicationFactorConfig. " +
+        "This error can be ignored if the cluster is starting up and not all brokers are up yet.")
       false
-    } else
+    } else {
       true
+    }
   }
 
-  private def metadataResponseTopic(error: Errors,
-                                    topic: String,
-                                    isInternal: Boolean,
-                                    partitionData: util.List[MetadataResponsePartition]): MetadataResponseTopic = {
-    new MetadataResponseTopic()
-      .setErrorCode(error.code)
-      .setName(topic)
-      .setIsInternal(isInternal)
-      .setPartitions(partitionData)
-  }
 }