Repository: apache/kafka
File: core/src/main/scala/kafka/server/AutoTopicCreationManager.scala (Scala)

Code before:
    metadataRequestContext: Option[RequestContext]
  ): Seq[MetadataResponseTopic] = {
    val (creatableTopics, uncreatableTopicResponses) = filterCreatableTopics(topics)

    val creatableTopicResponses = if (creatableTopics.isEmpty) {
      Seq.empty
    } else {
      sendCreateTopicRequest(creatableTopics, metadataRequestContext)
    }

    uncreatableTopicResponses ++ creatableTopicResponses
  }

  override def createStreamsInternalTopics(
    topics: Map[String, CreatableTopic],
    requestContext: RequestContext,
    timeoutMs: Long
  ): Unit = {
    if (topics.isEmpty) {
      return
    }

    val currentTimeMs = time.milliseconds()

    // Filter out topics that are in back-off (have cached errors)
    val cachedErrors = topicCreationErrorCache.getErrorsForTopics(topics.keySet, currentTimeMs)

    // Filter out topics that are:
    // 1. Already in error cache (back-off period)
    // 2. Already in-flight (concurrent request)
    val topicsToCreate = topics.filter { case (topicName, _) =>
      !cachedErrors.contains(topicName) &&  // Not in back-off
      inflightTopics.add(topicName)          // Not in-flight
    }

    if (topicsToCreate.nonEmpty) {
      sendCreateTopicRequestWithErrorCaching(topicsToCreate, Some(requestContext), timeoutMs)
    }
  }

  override def getStreamsInternalTopicCreationErrors(
    topicNames: Set[String],
    currentTimeMs: Long
  ): Map[String, String] = {
    topicCreationErrorCache.getErrorsForTopics(topicNames, currentTimeMs)
  }

  private def sendCreateTopicRequest(
    creatableTopics: Map[String, CreatableTopic],
    requestContext: Option[RequestContext]
  ): Seq[MetadataResponseTopic] = {

Review comment (performance):
That will create an extra collection and will be called fairly often. Maybe it would be better to implement a `topicCreationErrorCache.containsErrorsForTopic` and use this directly in the filter, to avoid the extra collection.

Code after:
   * @return auto created topic metadata responses
   */
  override def createTopics(
    topics: Set[String],
    controllerMutationQuota: ControllerMutationQuota,
    metadataRequestContext: Option[RequestContext]
  ): Seq[MetadataResponseTopic] = {
    val (creatableTopics, uncreatableTopicResponses) = filterCreatableTopics(topics)

    val creatableTopicResponses = if (creatableTopics.isEmpty) {
      Seq.empty
    } else {
      sendCreateTopicRequest(creatableTopics, metadataRequestContext)
    }

    uncreatableTopicResponses ++ creatableTopicResponses
  }

  override def createStreamsInternalTopics(
    topics: Map[String, CreatableTopic],
    requestContext: RequestContext,
    timeoutMs: Long
  ): Unit = {
    if (topics.isEmpty) {
      return
    }

    val currentTimeMs = time.milliseconds()

    // Filter out topics that are:
    // 1. Already in error cache (back-off period)
    // 2. Already in-flight (concurrent request)
    val topicsToCreate = topics.filter { case (topicName, _) =>
      !topicCreationErrorCache.hasError(topicName, currentTimeMs) &&
      inflightTopics.add(topicName)
    }

    if (topicsToCreate.nonEmpty) {
      sendCreateTopicRequestWithErrorCaching(topicsToCreate, Some(requestContext), timeoutMs)
    }
  }

  override def getStreamsInternalTopicCreationErrors(
    topicNames: Set[String],
    currentTimeMs: Long
  ): Map[String, String] = {
    topicCreationErrorCache.getErrorsForTopics(topicNames, currentTimeMs)
  }

  private def sendCreateTopicRequest(
    creatableTopics: Map[String, CreatableTopic],