Repository: apache/kafka
File: core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala (Scala)

Code before:

    val createTopicsResponse = new CreateTopicsResponse(createTopicsResponseData)
    val requestVersion = ApiKeys.CREATE_TOPICS.latestVersion()
    val correlationId = requestContext.correlationId  // Use the actual correlation ID from request context
    val clientId = requestContext.clientId

    // Serialize the CreateTopicsResponse with header
    val responseHeader = new ResponseHeader(correlationId, ApiKeys.CREATE_TOPICS.responseHeaderVersion(requestVersion))
    val serializedResponse = RequestUtils.serialize(responseHeader.data(), responseHeader.headerVersion(),
                                                     createTopicsResponse.data(), requestVersion)

    // Create an EnvelopeResponse containing the serialized CreateTopicsResponse
    val envelopeResponse = new EnvelopeResponse(serializedResponse, Errors.NONE)
    val requestHeader = new RequestHeader(ApiKeys.ENVELOPE, 0, clientId, correlationId)
    val clientResponse = new ClientResponse(requestHeader, null, null,
      0, 0, false, null, null, envelopeResponse)

    // Trigger the completion handler
    argumentCaptor.getValue.onComplete(clientResponse)

    // Verify only the failed topic was cached
    val cachedErrors = autoTopicCreationManager.getStreamsInternalTopicCreationErrors(
      Set("test-topic-1", "test-topic-2"), mockTime.milliseconds())
    
    // Debug output to understand the failure
    
    assertEquals(1, cachedErrors.size, s"Expected only 1 error but found: $cachedErrors")
    assertTrue(cachedErrors.contains("test-topic-2"))
    assertEquals("Topic already exists", cachedErrors("test-topic-2"))
  }

  @Test
  def testSendCreateTopicRequestEnvelopeHandling(): Unit = {
    // Test the sendCreateTopicRequest method (without error caching) handles envelopes correctly
    autoTopicCreationManager = new DefaultAutoTopicCreationManager(
      config,
      brokerToController,
      groupCoordinator,
      transactionCoordinator,
      shareCoordinator,
      mockTime,
      topicErrorCacheCapacity = testCacheCapacity)

    val requestContext = initializeRequestContextWithUserPrincipal()

    // Call createTopics which uses sendCreateTopicRequest internally
    autoTopicCreationManager.createTopics(
      Set("test-topic"), ControllerMutationQuota.UNBOUNDED_CONTROLLER_MUTATION_QUOTA, Some(requestContext))

    val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
    Mockito.verify(brokerToController).sendRequest(

Review comment (bug):
The empty debug comment on line 787 should be removed as it provides no value and appears to be leftover from development.
```suggestion

```

Code after:

    val createTopicsResponse = new CreateTopicsResponse(createTopicsResponseData)
    val requestVersion = ApiKeys.CREATE_TOPICS.latestVersion()
    val correlationId = requestContext.correlationId  // Use the actual correlation ID from request context
    val clientId = requestContext.clientId

    // Serialize the CreateTopicsResponse with header
    val responseHeader = new ResponseHeader(correlationId, ApiKeys.CREATE_TOPICS.responseHeaderVersion(requestVersion))
    val serializedResponse = RequestUtils.serialize(responseHeader.data(), responseHeader.headerVersion(),
                                                     createTopicsResponse.data(), requestVersion)

    // Create an EnvelopeResponse containing the serialized CreateTopicsResponse
    val envelopeResponse = new EnvelopeResponse(serializedResponse, Errors.NONE)
    val requestHeader = new RequestHeader(ApiKeys.ENVELOPE, 0, clientId, correlationId)
    val clientResponse = new ClientResponse(requestHeader, null, null,
      0, 0, false, null, null, envelopeResponse)

    // Trigger the completion handler
    argumentCaptor.getValue.onComplete(clientResponse)

    // Verify only the failed topic was cached
    val cachedErrors = autoTopicCreationManager.getStreamsInternalTopicCreationErrors(
      Set("test-topic-1", "test-topic-2"), mockTime.milliseconds())
    
    assertEquals(1, cachedErrors.size, s"Expected only 1 error but found: $cachedErrors")
    assertTrue(cachedErrors.contains("test-topic-2"))
    assertEquals("Topic already exists", cachedErrors("test-topic-2"))
  }

  @Test
  def testSendCreateTopicRequestEnvelopeHandling(): Unit = {
    // Test the sendCreateTopicRequest method (without error caching) handles envelopes correctly
    autoTopicCreationManager = new DefaultAutoTopicCreationManager(
      config,
      brokerToController,
      groupCoordinator,
      transactionCoordinator,
      shareCoordinator,
      mockTime,
      topicErrorCacheCapacity = testCacheCapacity)

    val requestContext = initializeRequestContextWithUserPrincipal()

    // Call createTopics which uses sendCreateTopicRequest internally
    autoTopicCreationManager.createTopics(
      Set("test-topic"), ControllerMutationQuota.UNBOUNDED_CONTROLLER_MUTATION_QUOTA, Some(requestContext))

    val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
    Mockito.verify(brokerToController).sendRequest(
      any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
      argumentCaptor.capture())