Fix a few lingering problems

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

@@ -17,7 +17,6 @@
 
 package kafka.server
 
-import java.util
 import java.util.Properties
 import java.util.concurrent.ConcurrentHashMap
 
@@ -31,13 +30,13 @@ 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}
 
 trait AutoTopicCreationManager {
 
@@ -80,12 +79,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 +94,7 @@ class AutoTopicCreationManagerImpl(
   txnCoordinator: TransactionCoordinator
 ) extends AutoTopicCreationManager with Logging {
 
-  private val inflightTopics = new ConcurrentHashMap[String, CreatableTopic]()
+  private val inflightTopics = new ConcurrentHashMap[String, Boolean]()
 
   override def start(): Unit = {
     channelManager.foreach(_.start())
@@ -106,28 +105,16 @@ 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
+  override def createTopics(
+    topics: Set[String],
+    controllerMutationQuota: ControllerMutationQuota
+  ): Seq[MetadataResponseTopic] = {
+    val (topicResponses, creatableTopics) = filterCreatableTopics(topics)
 
-    if (topicConfigs.nonEmpty) {
+    if (creatableTopics.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))
+        creatableTopics.foreach(config => topicsToCreate.add(config._2))
         val createTopicsRequest = new CreateTopicsRequest.Builder(
           new CreateTopicsRequestData()
             .setTimeoutMs(config.requestTimeoutMs)
@@ -137,40 +124,44 @@ class AutoTopicCreationManagerImpl(
         channelManager.get.sendRequest(createTopicsRequest, new ControllerRequestCompletionHandler {
           override def onTimeout(): Unit = {
             debug(s"Auto topic creation timed out for $topics.")
-            clearInflightRequests(topicConfigs)
+            clearInflightRequests(creatableTopics)
           }
 
           override def onComplete(response: ClientResponse): Unit = {
             debug(s"Auto topic creation completed for $topics.")
-            clearInflightRequests(topicConfigs)
+            clearInflightRequests(creatableTopics)
           }
         })
-        info(s"Sent $topics to the active controller for auto creation.")
+        info(s"Sent auto-creation request for $topics to the active controller.")
       } else {
-        adminManager.createTopics(
-          config.requestTimeoutMs,
-          validateOnly = false,
-          topicConfigs,
-          Map.empty,
-          controllerMutationQuota,
-          _ => ())
-        info(s"Topics $topics are being created asynchronously.")
+        try {
+          def creationCallback(topicErrors: Map[String, ApiError]): Unit = {
+            info(s"Auto-creation of topics $topics returned with errors: $topicErrors")
+          }
+
+          // Note that we use timeout = 0 since we do not need to wait for metadata propagation
+          adminManager.createTopics(
+            timeout = 0,
+            validateOnly = false,
+            creatableTopics,
+            Map.empty,
+            controllerMutationQuota,
+            creationCallback
+          )
+        } finally {
+          clearInflightRequests(creatableTopics)
+        }
       }
     }
-    topicResponses.toSeq
-  }
-
-  private def shouldCreate(topicResponse: MetadataResponseTopic): Boolean = {
-    Errors.forCode(topicResponse.errorCode()) == Errors.UNKNOWN_TOPIC_OR_PARTITION &&
-      !inflightTopics.containsKey(topicResponse.name)
+    topicResponses
   }
 
-  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 +195,56 @@ class AutoTopicCreationManagerImpl(
     topicConfigs
   }
 
-  private def hasEnoughAliveBrokers(topic: String): Boolean = {
-    if (topic == null)
-      throw new IllegalArgumentException("topic must not be null")
+  private def filterCreatableTopics(
+    topics: Set[String]
+  ): (Seq[MetadataResponseTopic], Map[String, CreatableTopic]) = {
 
     val aliveBrokers = metadataCache.getAliveBrokers
+    val creatableTopics = mutable.Map.empty[String, CreatableTopic]
+    val topicResponses = topics.toSeq.map { topic =>
+      val error = if (hasEnoughLiveBrokers(topic, aliveBrokers)) {
+        val alreadyPresent = inflightTopics.putIfAbsent(topic, true)
+        if (!alreadyPresent) {
+          creatableTopics.put(topic, creatableTopic(topic))
+        }
+
+        Errors.UNKNOWN_TOPIC_OR_PARTITION
+      } else {
+        Errors.INVALID_REPLICATION_FACTOR
+      }
 
-    topic match {
+      new MetadataResponseTopic()
+        .setErrorCode(error.code)
+        .setName(topic)
+        .setIsInternal(Topic.isInternal(topic))
+    }
+
+    (topicResponses, creatableTopics)
+  }
+
+  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)
-  }
 }

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

@@ -1101,20 +1101,36 @@ class KafkaApis(val requestChannel: RequestChannel,
       .setPartitions(partitionData)
   }
 
-  private def getTopicMetadata(isFetchAllMetadata: Boolean,
-                               topics: Set[String],
-                               listenerName: ListenerName,
-                               errorUnavailableEndpoints: Boolean,
-                               errorUnavailableListeners: Boolean): (Seq[MetadataResponseTopic], Set[String]) = {
+  private def getTopicMetadata(
+    request: RequestChannel.Request,
+    allowAutoTopicCreation: Boolean,
+    topics: Set[String],
+    listenerName: ListenerName,
+    errorUnavailableEndpoints: Boolean,
+    errorUnavailableListeners: Boolean
+  ): Seq[MetadataResponseTopic] = {
     val topicResponses = metadataCache.getTopicMetadata(topics, listenerName,
-        errorUnavailableEndpoints, errorUnavailableListeners)
+      errorUnavailableEndpoints, errorUnavailableListeners)
 
-    // A metadata request for all topics should never result in topic auto creation, but a topic may be deleted
-    // in between the creation of the topics parameter and topicResponses, so make sure to return None for this case.
-    if (isFetchAllMetadata || topics.isEmpty || topicResponses.size == topics.size) {
-      (topicResponses, Set.empty[String])
+    if (topics.isEmpty || topicResponses.size == topics.size) {
+      topicResponses
     } else {
-      (topicResponses, topics.diff(topicResponses.map(_.name).toSet))
+      val nonExistingTopics = topics.diff(topicResponses.map(_.name).toSet)
+      val nonExistingTopicResponses = if (allowAutoTopicCreation) {
+        val controllerMutationQuota = quotas.controllerMutation.newPermissiveQuotaFor(request)
+        autoTopicCreationManager.createTopics(nonExistingTopics, controllerMutationQuota)
+      } else {
+        nonExistingTopics.map { topic =>
+          metadataResponseTopic(
+            Errors.UNKNOWN_TOPIC_OR_PARTITION,
+            topic,
+            Topic.isInternal(topic),
+            util.Collections.emptyList()
+          )
+        }
+      }
+
+      topicResponses ++ nonExistingTopicResponses
     }
   }
 
@@ -1162,20 +1178,10 @@ class KafkaApis(val requestChannel: RequestChannel,
     // In versions 5 and below, we returned LEADER_NOT_AVAILABLE if a matching listener was not found on the leader.
     // From version 6 onwards, we return LISTENER_NOT_FOUND to enable diagnosis of configuration errors.
     val errorUnavailableListeners = requestVersion >= 6
-    val (topicMetadata, nonExistTopics) =
-      if (authorizedTopics.isEmpty)
-        (Seq.empty[MetadataResponseTopic], Set.empty[String])
-      else
-        getTopicMetadata(metadataRequest.isAllTopics, authorizedTopics,
-          request.context.listenerName, errorUnavailableEndpoints, errorUnavailableListeners)
 
-    val nonExistTopicMetadata =
-      if (nonExistTopics.nonEmpty && metadataRequest.allowAutoTopicCreation && config.autoCreateTopicsEnable) {
-        val controllerMutationQuota = quotas.controllerMutation.newQuotaFor(request, strictSinceVersion = 6)
-        autoTopicCreationManager.createTopics(
-          nonExistTopics.toSet, controllerMutationQuota)
-      } else
-        Seq.empty[MetadataResponseTopic]
+    val allowAutoCreation = metadataRequest.allowAutoTopicCreation && !metadataRequest.isAllTopics
+    val topicMetadata = getTopicMetadata(request, allowAutoCreation, authorizedTopics,
+      request.context.listenerName, errorUnavailableEndpoints, errorUnavailableListeners)
 
     var clusterAuthorizedOperations = Int.MinValue // Default value in the schema
     if (requestVersion >= 8) {
@@ -1197,11 +1203,10 @@ class KafkaApis(val requestChannel: RequestChannel,
           }
         }
         setTopicAuthorizedOperations(topicMetadata)
-        setTopicAuthorizedOperations(nonExistTopicMetadata)
       }
     }
 
-    val completeTopicMetadata = topicMetadata ++ nonExistTopicMetadata ++ unauthorizedForCreateTopicMetadata ++ unauthorizedForDescribeTopicMetadata
+    val completeTopicMetadata = topicMetadata ++ unauthorizedForCreateTopicMetadata ++ unauthorizedForDescribeTopicMetadata
 
     val brokers = metadataCache.getAliveBrokers
 
@@ -1331,7 +1336,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       }
 
       if (topicMetadata.headOption.isEmpty) {
-        val controllerMutationQuota = quotas.controllerMutation.newQuotaFor(request, strictSinceVersion = 6)
+        val controllerMutationQuota = quotas.controllerMutation.newPermissiveQuotaFor(request)
         autoTopicCreationManager.createTopics(Seq(internalTopicName).toSet, controllerMutationQuota)
         requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => createFindCoordinatorResponse(
           Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode, requestThrottleMs))

core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala

@@ -92,7 +92,7 @@ class AutoTopicCreationManagerTest {
                               isInternal: Boolean,
                               numPartitions: Int = 1,
                               replicationFactor: Short = 1): Unit = {
-    autoTopicCreationManager = new AutoTopicCreationManagerImpl(
+    autoTopicCreationManager = new DefaultAutoTopicCreationManager(
       config,
       metadataCache,
       Some(brokerToController),
@@ -121,7 +121,7 @@ class AutoTopicCreationManagerTest {
 
   @Test
   def testCreateTopicsWithForwardingDisabled(): Unit = {
-    autoTopicCreationManager = new AutoTopicCreationManagerImpl(
+    autoTopicCreationManager = new DefaultAutoTopicCreationManager(
       config,
       metadataCache,
       None,
@@ -137,7 +137,7 @@ class AutoTopicCreationManagerTest {
     createTopicAndVerifyResult(Errors.UNKNOWN_TOPIC_OR_PARTITION, topicName, false)
 
     Mockito.verify(adminManager).createTopics(
-      ArgumentMatchers.eq(requestTimeout),
+      ArgumentMatchers.eq(0),
       ArgumentMatchers.eq(false),
       ArgumentMatchers.eq(Map(topicName -> getNewTopic(topicName))),
       ArgumentMatchers.eq(Map.empty),
@@ -151,7 +151,7 @@ class AutoTopicCreationManagerTest {
     props.setProperty(KafkaConfig.DefaultReplicationFactorProp, 3.toString)
     config = KafkaConfig.fromProps(props)
 
-    autoTopicCreationManager = new AutoTopicCreationManagerImpl(
+    autoTopicCreationManager = new DefaultAutoTopicCreationManager(
       config,
       metadataCache,
       Some(brokerToController),

core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala

@@ -163,8 +163,8 @@ class MetadataRequestTest extends AbstractMetadataRequestTest {
 
   @Test
   def testAutoCreateOfCollidingTopics(): Unit = {
-    val topic1 = "testAutoCreate_Topic"
-    val topic2 = "testAutoCreate.Topic"
+    val topic1 = "testAutoCreate.Topic"
+    val topic2 = "testAutoCreate_Topic"
     val response1 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1, topic2).asJava, true).build)
     assertEquals(2, response1.topicMetadata.size)
     var topicMetadata1 = response1.topicMetadata.asScala.head

core/src/test/scala/unit/kafka/server/MetadataRequestWithForwardingTest.scala

@@ -81,8 +81,8 @@ class MetadataRequestWithForwardingTest extends AbstractMetadataRequestTest {
 
   @Test
   def testAutoCreateOfCollidingTopics(): Unit = {
-    val topic1 = "testAutoCreate_Topic"
-    val topic2 = "testAutoCreate.Topic"
+    val topic1 = "testAutoCreate.Topic"
+    val topic2 = "testAutoCreate_Topic"
     val response1 = sendMetadataRequest(new MetadataRequest.Builder(Seq(topic1, topic2).asJava, true).build)
     assertEquals(2, response1.topicMetadata.size)
     var topicMetadata1 = response1.topicMetadata.asScala.head