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

Code before:
                  requestHeaderForParsing match {
                    case Some(requestHeader) =>
                      try {
                        // Use the captured request header for proper envelope response parsing
                        val createTopicsResponse = AbstractResponse.parseResponse(
                          envelopeResponse.responseData(), requestHeader).asInstanceOf[CreateTopicsResponse]

                        cacheTopicCreationErrorsFromResponse(createTopicsResponse, timeoutMs)
                      } catch {
                        case e: Exception =>
                          warn(s"Failed to parse envelope response for auto topic creation of ${creatableTopics.keys}", e)
                          cacheTopicCreationErrors(creatableTopics.keys.toSet, s"Response parsing error: ${e.getMessage}", timeoutMs)
                      }
                    case None =>
                      warn(s"Cannot parse envelope response without original request header information")
                      cacheTopicCreationErrors(creatableTopics.keys.toSet, "Missing request header for envelope parsing", timeoutMs)
                  }
                }
              case createTopicsResponse: CreateTopicsResponse =>
                cacheTopicCreationErrorsFromResponse(createTopicsResponse, timeoutMs)
              case other =>
                warn(s"Auto topic creation request received unexpected response type: ${other.getClass.getSimpleName}")
                cacheTopicCreationErrors(creatableTopics.keys.toSet, s"Unexpected response type: ${other.getClass.getSimpleName}", timeoutMs)
            }
          }
          debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
        }
      }
    }

    val request = (requestContext, requestHeaderForParsing) match {
      case (Some(context), Some(requestHeader)) =>
        ForwardingManager.buildEnvelopeRequest(context,
          createTopicsRequest.build(requestHeader.apiVersion()).serializeWithHeader(requestHeader))
      case _ =>
        createTopicsRequest
    }

    channelManager.sendRequest(request, requestCompletionHandler)

    val creatableTopicResponses = creatableTopics.keySet.toSeq.map { topic =>
      new MetadataResponseTopic()
        .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
        .setName(topic)
        .setIsInternal(Topic.isInternal(topic))
    }

    creatableTopicResponses
  }

  private def cacheTopicCreationErrors(topicNames: Set[String], errorMessage: String, ttlMs: Long): Unit = {

Review comment (bug):
This debug statement is placed outside the response handling conditional block, causing it to execute even when there's no response body. It should be moved inside the `if (response.hasResponse)` block to avoid logging when no response is available.
```suggestion
            debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
          }
```

Code after:
                  requestHeaderForParsing match {
                    case Some(requestHeader) =>
                      try {
                        // Use the captured request header for proper envelope response parsing
                        val createTopicsResponse = AbstractResponse.parseResponse(
                          envelopeResponse.responseData(), requestHeader).asInstanceOf[CreateTopicsResponse]

                        cacheTopicCreationErrorsFromResponse(createTopicsResponse, timeoutMs)
                      } catch {
                        case e: Exception =>
                          warn(s"Failed to parse envelope response for auto topic creation of ${creatableTopics.keys}", e)
                          cacheTopicCreationErrors(creatableTopics.keys.toSet, s"Response parsing error: ${e.getMessage}", timeoutMs)
                      }
                    case None =>
                      warn(s"Cannot parse envelope response without original request header information")
                      cacheTopicCreationErrors(creatableTopics.keys.toSet, "Missing request header for envelope parsing", timeoutMs)
                  }
                }
              case createTopicsResponse: CreateTopicsResponse =>
                cacheTopicCreationErrorsFromResponse(createTopicsResponse, timeoutMs)
              case unexpectedResponse =>
                warn(s"Auto topic creation request received unexpected response type: ${unexpectedResponse.getClass.getSimpleName}")
                cacheTopicCreationErrors(creatableTopics.keys.toSet, s"Unexpected response type: ${unexpectedResponse.getClass.getSimpleName}", timeoutMs)
            }
            debug(s"Auto topic creation completed for ${creatableTopics.keys} with response ${response.responseBody}.")
          }
        }
      }
    }

    val request = (requestContext, requestHeaderForParsing) match {
      case (Some(context), Some(requestHeader)) =>
        ForwardingManager.buildEnvelopeRequest(context,
          createTopicsRequest.build(requestHeader.apiVersion()).serializeWithHeader(requestHeader))
      case _ =>
        createTopicsRequest
    }

    channelManager.sendRequest(request, requestCompletionHandler)

    val creatableTopicResponses = creatableTopics.keySet.toSeq.map { topic =>
      new MetadataResponseTopic()
        .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
        .setName(topic)
        .setIsInternal(Topic.isInternal(topic))
    }

    creatableTopicResponses
  }

  private def cacheTopicCreationErrors(topicNames: Set[String], errorMessage: String, ttlMs: Long): Unit = {