You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2025/09/24 09:14:32 UTC
Re: [PR] MINOR: Handle envelope response in AutoTopicCreationManager [kafka]
Copilot commented on code in PR #20569:
URL: https://github.com/apache/kafka/pull/20569#discussion_r2375132004
##########
core/src/main/scala/kafka/server/AutoTopicCreationManager.scala:
##########
@@ -229,26 +272,13 @@ class DefaultAutoTopicCreationManager(
}
}
- 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,
- requestVersion,
- context.clientId,
- context.correlationId)
- ForwardingManager.buildEnvelopeRequest(context,
- createTopicsRequest.build(requestVersion).serializeWithHeader(requestHeader))
- }.getOrElse(createTopicsRequest)
+ val request = (requestContext, requestHeaderForParsing) match {
+ case (Some(context), Some(requestHeader)) =>
+ ForwardingManager.buildEnvelopeRequest(context,
+ createTopicsRequest.build(requestHeader.apiVersion()).serializeWithHeader(requestHeader))
+ case _ =>
+ createTopicsRequest
+ }
Review Comment:
The pattern matching logic is duplicated in two locations (lines 275-281 and 470-476). Consider extracting this into a helper method to avoid code duplication.
##########
core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala:
##########
@@ -523,6 +524,326 @@ class AutoTopicCreationManagerTest {
assertTrue(expiredErrors.isEmpty, "Expired errors should be proactively cleaned up")
}
+ @Test
+ def testEnvelopeResponseSuccessfulParsing(): Unit = {
+ autoTopicCreationManager = new DefaultAutoTopicCreationManager(
+ config,
+ brokerToController,
+ groupCoordinator,
+ transactionCoordinator,
+ shareCoordinator,
+ mockTime,
+ topicErrorCacheCapacity = testCacheCapacity)
+
+ val topics = Map(
+ "test-topic" -> new CreatableTopic().setName("test-topic").setNumPartitions(1).setReplicationFactor(1)
+ )
+ val requestContext = initializeRequestContextWithUserPrincipal()
+ val timeoutMs = 5000L
+
+ autoTopicCreationManager.createStreamsInternalTopics(topics, requestContext, timeoutMs)
+
+ val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
+ Mockito.verify(brokerToController).sendRequest(
+ any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
+ argumentCaptor.capture())
+
+ // Create a successful CreateTopicsResponse
+ val createTopicsResponseData = new org.apache.kafka.common.message.CreateTopicsResponseData()
+ val topicResult = new org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult()
+ .setName("test-topic")
+ .setErrorCode(Errors.NONE.code())
+ .setNumPartitions(1)
+ .setReplicationFactor(1.toShort)
+ createTopicsResponseData.topics().add(topicResult)
+
+ 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 as it would appear in an envelope
+ 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 no errors were cached (successful response)
+ val cachedErrors = autoTopicCreationManager.getStreamsInternalTopicCreationErrors(Set("test-topic"), mockTime.milliseconds())
+ assertTrue(cachedErrors.isEmpty, "No errors should be cached for successful response")
+ }
+
+ @Test
+ def testEnvelopeResponseWithEnvelopeError(): Unit = {
+ autoTopicCreationManager = new DefaultAutoTopicCreationManager(
+ config,
+ brokerToController,
+ groupCoordinator,
+ transactionCoordinator,
+ shareCoordinator,
+ mockTime,
+ topicErrorCacheCapacity = testCacheCapacity)
+
+ val topics = Map(
+ "test-topic" -> new CreatableTopic().setName("test-topic").setNumPartitions(1).setReplicationFactor(1)
+ )
+ val requestContext = initializeRequestContextWithUserPrincipal()
+ val timeoutMs = 5000L
+
+ autoTopicCreationManager.createStreamsInternalTopics(topics, requestContext, timeoutMs)
+
+ val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
+ Mockito.verify(brokerToController).sendRequest(
+ any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
+ argumentCaptor.capture())
+
+ // Create an EnvelopeResponse with an envelope-level error
+ val envelopeResponse = new EnvelopeResponse(ByteBuffer.allocate(0), Errors.UNSUPPORTED_VERSION)
+ val requestHeader = new RequestHeader(ApiKeys.ENVELOPE, 0, requestContext.clientId, requestContext.correlationId)
+ val clientResponse = new ClientResponse(requestHeader, null, null,
+ 0, 0, false, null, null, envelopeResponse)
+
+ // Trigger the completion handler
+ argumentCaptor.getValue.onComplete(clientResponse)
+
+ // Verify the envelope error was cached
+ val cachedErrors = autoTopicCreationManager.getStreamsInternalTopicCreationErrors(Set("test-topic"), mockTime.milliseconds())
+ assertEquals(1, cachedErrors.size)
+ assertTrue(cachedErrors("test-topic").contains("Envelope error: UNSUPPORTED_VERSION"))
+ }
+
+ @Test
+ def testEnvelopeResponseParsingException(): Unit = {
+ autoTopicCreationManager = new DefaultAutoTopicCreationManager(
+ config,
+ brokerToController,
+ groupCoordinator,
+ transactionCoordinator,
+ shareCoordinator,
+ mockTime,
+ topicErrorCacheCapacity = testCacheCapacity)
+
+ val topics = Map(
+ "test-topic" -> new CreatableTopic().setName("test-topic").setNumPartitions(1).setReplicationFactor(1)
+ )
+ val requestContext = initializeRequestContextWithUserPrincipal()
+ val timeoutMs = 5000L
+
+ autoTopicCreationManager.createStreamsInternalTopics(topics, requestContext, timeoutMs)
+
+ val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
+ Mockito.verify(brokerToController).sendRequest(
+ any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
+ argumentCaptor.capture())
+
+ // Create an EnvelopeResponse with malformed response data that will cause parsing to fail
+ val malformedData = ByteBuffer.wrap("invalid response data".getBytes())
+ val envelopeResponse = new EnvelopeResponse(malformedData, Errors.NONE)
+ val requestHeader = new RequestHeader(ApiKeys.ENVELOPE, 0, requestContext.clientId, requestContext.correlationId)
+ val clientResponse = new ClientResponse(requestHeader, null, null,
+ 0, 0, false, null, null, envelopeResponse)
+
+ // Trigger the completion handler
+ argumentCaptor.getValue.onComplete(clientResponse)
+
+ // Verify the parsing error was cached
+ val cachedErrors = autoTopicCreationManager.getStreamsInternalTopicCreationErrors(Set("test-topic"), mockTime.milliseconds())
+ assertEquals(1, cachedErrors.size)
+ assertTrue(cachedErrors("test-topic").contains("Response parsing error:"))
+ }
+
+ @Test
+ def testEnvelopeResponseCorrelationIdMismatch(): Unit = {
+ autoTopicCreationManager = new DefaultAutoTopicCreationManager(
+ config,
+ brokerToController,
+ groupCoordinator,
+ transactionCoordinator,
+ shareCoordinator,
+ mockTime,
+ topicErrorCacheCapacity = testCacheCapacity)
+
+ val topics = Map(
+ "test-topic" -> new CreatableTopic().setName("test-topic").setNumPartitions(1).setReplicationFactor(1)
+ )
+ val requestContext = initializeRequestContextWithUserPrincipal()
+ val timeoutMs = 5000L
+
+ autoTopicCreationManager.createStreamsInternalTopics(topics, requestContext, timeoutMs)
+
+ val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
+ Mockito.verify(brokerToController).sendRequest(
+ any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
+ argumentCaptor.capture())
+
+ // Create a CreateTopicsResponse with a different correlation ID than the request
+ val createTopicsResponseData = new org.apache.kafka.common.message.CreateTopicsResponseData()
+ val topicResult = new org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult()
+ .setName("test-topic")
+ .setErrorCode(Errors.NONE.code())
+ createTopicsResponseData.topics().add(topicResult)
+
+ val createTopicsResponse = new CreateTopicsResponse(createTopicsResponseData)
+ val requestVersion = ApiKeys.CREATE_TOPICS.latestVersion()
+ val requestCorrelationId = 123
+ val responseCorrelationId = 456 // Different correlation ID
+ val clientId = "test-client"
+
+ // Serialize the CreateTopicsResponse with mismatched correlation ID
+ val responseHeader = new ResponseHeader(responseCorrelationId, 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, requestCorrelationId)
+ val clientResponse = new ClientResponse(requestHeader, null, null,
+ 0, 0, false, null, null, envelopeResponse)
+
+ // Trigger the completion handler
+ argumentCaptor.getValue.onComplete(clientResponse)
+
+ // Verify the correlation ID mismatch error was cached
+ val cachedErrors = autoTopicCreationManager.getStreamsInternalTopicCreationErrors(Set("test-topic"), mockTime.milliseconds())
+ assertEquals(1, cachedErrors.size)
+ assertTrue(cachedErrors("test-topic").contains("Response parsing error:"))
+ }
+
+ @Test
+ def testEnvelopeResponseWithTopicErrors(): Unit = {
+ autoTopicCreationManager = new DefaultAutoTopicCreationManager(
+ config,
+ brokerToController,
+ groupCoordinator,
+ transactionCoordinator,
+ shareCoordinator,
+ mockTime,
+ topicErrorCacheCapacity = testCacheCapacity)
+
+ val topics = Map(
+ "test-topic-1" -> new CreatableTopic().setName("test-topic-1").setNumPartitions(1).setReplicationFactor(1),
+ "test-topic-2" -> new CreatableTopic().setName("test-topic-2").setNumPartitions(1).setReplicationFactor(1)
+ )
+ val requestContext = initializeRequestContextWithUserPrincipal()
+ val timeoutMs = 5000L
+
+ autoTopicCreationManager.createStreamsInternalTopics(topics, requestContext, timeoutMs)
+
+ val argumentCaptor = ArgumentCaptor.forClass(classOf[ControllerRequestCompletionHandler])
+ Mockito.verify(brokerToController).sendRequest(
+ any(classOf[AbstractRequest.Builder[_ <: AbstractRequest]]),
+ argumentCaptor.capture())
+
+ // Create a CreateTopicsResponse with mixed success and error results
+ val createTopicsResponseData = new org.apache.kafka.common.message.CreateTopicsResponseData()
+
+ // Successful topic
+ val successResult = new org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult()
+ .setName("test-topic-1")
+ .setErrorCode(Errors.NONE.code())
+ .setNumPartitions(1)
+ .setReplicationFactor(1.toShort)
+ createTopicsResponseData.topics().add(successResult)
+
+ // Failed topic
+ val errorResult = new org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult()
+ .setName("test-topic-2")
+ .setErrorCode(Errors.TOPIC_ALREADY_EXISTS.code())
+ .setErrorMessage("Topic already exists")
+ createTopicsResponseData.topics().add(errorResult)
+
+ 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
+
Review Comment:
The empty debug comment on line 787 should be removed as it provides no value and appears to be leftover from development.
```suggestion
```
##########
core/src/main/scala/kafka/server/AutoTopicCreationManager.scala:
##########
@@ -364,6 +394,22 @@ class DefaultAutoTopicCreationManager(
.setTopics(topicsToCreate)
)
+ // Capture request header information for proper envelope response parsing
+ val requestHeaderForParsing = requestContext.map { context =>
+ val requestVersion =
+ channelManager.controllerApiVersions.toScala match {
+ case None =>
+ ApiKeys.CREATE_TOPICS.latestVersion()
+ case Some(nodeApiVersions) =>
+ nodeApiVersions.latestUsableVersion(ApiKeys.CREATE_TOPICS)
+ }
+
+ new RequestHeader(ApiKeys.CREATE_TOPICS,
+ requestVersion,
+ context.clientId,
+ context.correlationId)
+ }
Review Comment:
The request header creation logic is duplicated between lines 202-215 and 398-411. Consider extracting this into a helper method to reduce code duplication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org