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

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

    val requestCompletionHandler = new ControllerRequestCompletionHandler {
      override def onTimeout(): Unit = {
        clearInflightRequests(creatableTopics)
        debug(s"Auto topic creation timed out for ${creatableTopics.keys}.")
      }

      override def onComplete(response: ClientResponse): Unit = {
        clearInflightRequests(creatableTopics)
        if (response.authenticationException() != null) {
          warn(s"Auto topic creation failed for ${creatableTopics.keys} with authentication exception")
        } else if (response.versionMismatch() != null) {
          warn(s"Auto topic creation failed for ${creatableTopics.keys} with invalid version exception")
        } else {
          if (response.hasResponse) {
            response.responseBody() match {
              case createTopicsResponse: CreateTopicsResponse =>
                createTopicsResponse.data().topics().forEach(topicResult => {
                  val error = Errors.forCode(topicResult.errorCode())
                  if (error != Errors.NONE) {
                    warn(s"Auto topic creation failed for ${topicResult.name()} with error '${error.name()}': ${topicResult.errorMessage()}")
                  }
                })
              case other =>
                warn(s"Auto topic creation request received unexpected response type: ${other.getClass.getSimpleName}")
            }
          }
          debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
        }
      }
    }

    val request = requestContext.map { context =>
      val requestVersion =
        channelManager.controllerApiVersions.toScala match {
          case None =>
            // We will rely on the Metadata request to be retried in the case
            // that the latest version is not usable by the controller.
            ApiKeys.CREATE_TOPICS.latestVersion()
          case Some(nodeApiVersions) =>
            nodeApiVersions.latestUsableVersion(ApiKeys.CREATE_TOPICS)
        }

      // Borrow client information such as client id and correlation id from the original request,
      // in order to correlate the create request with the original metadata request.
      val requestHeader = new RequestHeader(ApiKeys.CREATE_TOPICS,

Review comment (bug):
tiny nit: we can skip `()` for getters
```suggestion
                    warn(s"Auto topic creation failed for ${topicResult.name} with error '${error.name}': ${topicResult.errorMessage}")
```

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

    val requestCompletionHandler = new ControllerRequestCompletionHandler {
      override def onTimeout(): Unit = {
        clearInflightRequests(creatableTopics)
        debug(s"Auto topic creation timed out for ${creatableTopics.keys}.")
      }

      override def onComplete(response: ClientResponse): Unit = {
        clearInflightRequests(creatableTopics)
        if (response.authenticationException() != null) {
          warn(s"Auto topic creation failed for ${creatableTopics.keys} with authentication exception")
        } else if (response.versionMismatch() != null) {
          warn(s"Auto topic creation failed for ${creatableTopics.keys} with invalid version exception")
        } else {
          if (response.hasResponse) {
            response.responseBody() match {
              case createTopicsResponse: CreateTopicsResponse =>
                createTopicsResponse.data().topics().forEach(topicResult => {
                  val error = Errors.forCode(topicResult.errorCode)
                  if (error != Errors.NONE) {
                    warn(s"Auto topic creation failed for ${topicResult.name} with error '${error.name}': ${topicResult.errorMessage}")
                  }
                })
              case other =>
                warn(s"Auto topic creation request received unexpected response type: ${other.getClass.getSimpleName}")
            }
          }
          debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
        }
      }
    }

    val request = requestContext.map { context =>
      val requestVersion =
        channelManager.controllerApiVersions.toScala match {
          case None =>
            // We will rely on the Metadata request to be retried in the case
            // that the latest version is not usable by the controller.
            ApiKeys.CREATE_TOPICS.latestVersion()
          case Some(nodeApiVersions) =>
            nodeApiVersions.latestUsableVersion(ApiKeys.CREATE_TOPICS)
        }

      // Borrow client information such as client id and correlation id from the original request,
      // in order to correlate the create request with the original metadata request.
      val requestHeader = new RequestHeader(ApiKeys.CREATE_TOPICS,