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

Code before:
      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] = {
    val topicsToCreate = new CreateTopicsRequestData.CreatableTopicCollection(creatableTopics.size)
    topicsToCreate.addAll(creatableTopics.values.asJavaCollection)

    val createTopicsRequest = new CreateTopicsRequest.Builder(
      new CreateTopicsRequestData()
        .setTimeoutMs(config.requestTimeoutMs)
        .setTopics(topicsToCreate)

Review comment (suggestion):
Did we not add to inflightTopics before? Seems like it.

I think this is correct, but the inline comment is misaligned (I would just remove the inline comment). If you want to keep the inline comment above, I would add that you are actually adding this to inflight topics here.

Code after:
    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],
    requestContext: Option[RequestContext]
  ): Seq[MetadataResponseTopic] = {
    val topicsToCreate = new CreateTopicsRequestData.CreatableTopicCollection(creatableTopics.size)
    topicsToCreate.addAll(creatableTopics.values.asJavaCollection)

    val createTopicsRequest = new CreateTopicsRequest.Builder(
      new CreateTopicsRequestData()