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

Code before:
    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)
    )

    // Capture request header information for proper envelope response parsing
    val requestHeaderForParsing = requestContext.map { context =>
      val requestVersion =
        channelManager.controllerApiVersions.toScala match {
          case None =>

Review comment (performance):
[nitpick] Early return pattern breaks the conventional Scala style. Consider using an if-else structure or pattern matching instead of explicit return statements.
```suggestion
    if (topics.nonEmpty) {
      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)
      }
    }
```

Code after:

    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()
        .setTimeoutMs(config.requestTimeoutMs)
        .setTopics(topicsToCreate)
    )

    // Capture request header information for proper envelope response parsing
    val requestHeaderForParsing = requestContext.map { context =>
      val requestVersion =