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

Code before:

    // 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
    if (cachedErrors.size != 1) {
      println(s"Expected 1 cached error but found ${cachedErrors.size}")
      cachedErrors.foreach { case (topic, error) =>
        println(s"  Cached error for $topic: $error")
      }
    }
    
    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])

Review comment (bug):
Debug print statements should not be left in production test code. Consider removing these println statements or replacing them with proper logging if debugging information is needed.
```suggestion

```

Code after:

    // 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())

    // Create a CreateTopicsResponse with an error
    val createTopicsResponseData = new org.apache.kafka.common.message.CreateTopicsResponseData()
    val topicResult = new org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult()
      .setName("test-topic")