From f667edc6696f43a3bd9c751b3b8ce2e689c1fc8d Mon Sep 17 00:00:00 2001 From: TaiJu Wu Date: Wed, 8 Jan 2025 12:21:37 +0000 Subject: [PATCH 1/2] remove envelop --- .../main/scala/kafka/server/KafkaApis.scala | 55 +----- .../unit/kafka/server/KafkaApisTest.scala | 163 +----------------- 2 files changed, 5 insertions(+), 213 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index c1e8807de9d68..87423bb0d7f7a 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -91,7 +91,6 @@ import scala.annotation.nowarn import scala.collection.mutable.ArrayBuffer import scala.collection.{Map, Seq, Set, mutable} import scala.jdk.CollectionConverters._ -import scala.util.{Failure, Success, Try} /** * Logic to handle the various Kafka requests @@ -138,10 +137,6 @@ class KafkaApis(val requestChannel: RequestChannel, info("Shutdown complete.") } - private def isForwardingEnabled(request: RequestChannel.Request): Boolean = { - metadataSupport.forwardingManager.isDefined && request.context.principalSerde.isPresent - } - private def maybeForwardToController( request: RequestChannel.Request, handler: RequestChannel.Request => Unit @@ -249,7 +244,7 @@ class KafkaApis(val requestChannel: RequestChannel, case ApiKeys.ALTER_USER_SCRAM_CREDENTIALS => maybeForwardToController(request, handleAlterUserScramCredentialsRequest) case ApiKeys.ALTER_PARTITION => handleAlterPartitionRequest(request) case ApiKeys.UPDATE_FEATURES => maybeForwardToController(request, handleUpdateFeatures) - case ApiKeys.ENVELOPE => handleEnvelope(request, requestLocal) + case ApiKeys.ENVELOPE => handleEnvelope(request) case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request) case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request) case ApiKeys.UNREGISTER_BROKER => forwardToControllerOrFail(request) @@ -425,24 +420,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleControlledShutdownRequest(request: RequestChannel.Request): Unit = { - val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldNeverReceive(request)) - // ensureTopicExists is only for client facing requests - // We can't have the ensureTopicExists check here since the controller sends it as an advisory to all brokers so they - // stop serving data to clients for the topic being deleted - val controlledShutdownRequest = request.body[ControlledShutdownRequest] - authHelper.authorizeClusterOperation(request, CLUSTER_ACTION) - - def controlledShutdownCallback(controlledShutdownResult: Try[Set[TopicPartition]]): Unit = { - val response = controlledShutdownResult match { - case Success(partitionsRemaining) => - ControlledShutdownResponse.prepareResponse(Errors.NONE, partitionsRemaining.asJava) - - case Failure(throwable) => - controlledShutdownRequest.getErrorResponse(throwable) - } - requestHelper.sendResponseExemptThrottle(request, response) - } - zkSupport.controller.controlledShutdown(controlledShutdownRequest.data.brokerId, controlledShutdownRequest.data.brokerEpoch, controlledShutdownCallback) + throw KafkaApis.shouldNeverReceive(request) } /** @@ -3482,33 +3460,8 @@ class KafkaApis(val requestChannel: RequestChannel, new DescribeClusterResponse(response.setThrottleTimeMs(requestThrottleMs))) } - def handleEnvelope(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = { - val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldNeverReceive(request)) - - // If forwarding is not yet enabled or this request has been received on an invalid endpoint, - // then we treat the request as unparsable and close the connection. - if (!isForwardingEnabled(request)) { - info(s"Closing connection ${request.context.connectionId} because it sent an `Envelope` " + - "request even though forwarding has not been enabled") - requestChannel.closeConnection(request, Collections.emptyMap()) - return - } else if (!request.context.fromPrivilegedListener) { - info(s"Closing connection ${request.context.connectionId} from listener ${request.context.listenerName} " + - s"because it sent an `Envelope` request, which is only accepted on the inter-broker listener " + - s"${config.interBrokerListenerName}.") - requestChannel.closeConnection(request, Collections.emptyMap()) - return - } else if (!authHelper.authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME)) { - requestHelper.sendErrorResponseMaybeThrottle(request, new ClusterAuthorizationException( - s"Principal ${request.context.principal} does not have required CLUSTER_ACTION for envelope")) - return - } else if (!zkSupport.controller.isActive) { - requestHelper.sendErrorResponseMaybeThrottle(request, new NotControllerException( - s"Broker $brokerId is not the active controller")) - return - } - - EnvelopeUtils.handleEnvelopeRequest(request, requestChannel.metrics, handle(_, requestLocal)) + def handleEnvelope(request: RequestChannel.Request): Unit = { + throw KafkaApis.shouldNeverReceive(request) } def handleDescribeProducersRequest(request: RequestChannel.Request): Unit = { diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala index 229b1a18a19a7..387f168b8fa6f 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala @@ -324,167 +324,6 @@ class KafkaApisTest extends Logging { assertEquals(propValue, describeConfigsResponseData.value) } - @Test - def testEnvelopeRequestHandlingAsController(): Unit = { - testEnvelopeRequestWithAlterConfig( - alterConfigHandler = () => ApiError.NONE, - expectedError = Errors.NONE - ) - } - - @Test - def testEnvelopeRequestWithAlterConfigUnhandledError(): Unit = { - testEnvelopeRequestWithAlterConfig( - alterConfigHandler = () => throw new IllegalStateException(), - expectedError = Errors.UNKNOWN_SERVER_ERROR - ) - } - - private def testEnvelopeRequestWithAlterConfig( - alterConfigHandler: () => ApiError, - expectedError: Errors - ): Unit = { - val authorizer: Authorizer = mock(classOf[Authorizer]) - - authorizeResource(authorizer, AclOperation.CLUSTER_ACTION, ResourceType.CLUSTER, Resource.CLUSTER_NAME, AuthorizationResult.ALLOWED) - - val operation = AclOperation.ALTER_CONFIGS - val resourceName = "topic-1" - val requestHeader = new RequestHeader(ApiKeys.ALTER_CONFIGS, ApiKeys.ALTER_CONFIGS.latestVersion, - clientId, 0) - - when(controller.isActive).thenReturn(true) - - authorizeResource(authorizer, operation, ResourceType.TOPIC, resourceName, AuthorizationResult.ALLOWED) - - val configResource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName) - when(adminManager.alterConfigs(any(), ArgumentMatchers.eq(false))) - .thenAnswer(_ => { - Map(configResource -> alterConfigHandler.apply()) - }) - - val configs = Map( - configResource -> new AlterConfigsRequest.Config( - Seq(new AlterConfigsRequest.ConfigEntry("foo", "bar")).asJava)) - val alterConfigsRequest = new AlterConfigsRequest.Builder(configs.asJava, false).build(requestHeader.apiVersion) - - val startTimeNanos = time.nanoseconds() - val queueDurationNanos = 5 * 1000 * 1000 - val request = TestUtils.buildEnvelopeRequest( - alterConfigsRequest, kafkaPrincipalSerde, requestChannelMetrics, startTimeNanos, startTimeNanos + queueDurationNanos) - - val capturedResponse: ArgumentCaptor[AlterConfigsResponse] = ArgumentCaptor.forClass(classOf[AlterConfigsResponse]) - val capturedRequest: ArgumentCaptor[RequestChannel.Request] = ArgumentCaptor.forClass(classOf[RequestChannel.Request]) - kafkaApis = createKafkaApis(authorizer = Some(authorizer), enableForwarding = true) - kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching) - - verify(requestChannel).sendResponse( - capturedRequest.capture(), - capturedResponse.capture(), - any() - ) - assertEquals(Some(request), capturedRequest.getValue.envelope) - // the dequeue time of forwarded request should equals to envelop request - assertEquals(request.requestDequeueTimeNanos, capturedRequest.getValue.requestDequeueTimeNanos) - val innerResponse = capturedResponse.getValue - val responseMap = innerResponse.data.responses().asScala.map { resourceResponse => - resourceResponse.resourceName -> Errors.forCode(resourceResponse.errorCode) - }.toMap - - assertEquals(Map(resourceName -> expectedError), responseMap) - - verify(controller).isActive - verify(adminManager).alterConfigs(any(), ArgumentMatchers.eq(false)) - } - - @Test - def testInvalidEnvelopeRequestWithNonForwardableAPI(): Unit = { - val requestHeader = new RequestHeader(ApiKeys.LEAVE_GROUP, ApiKeys.LEAVE_GROUP.latestVersion, - clientId, 0) - val leaveGroupRequest = new LeaveGroupRequest.Builder("group", - Collections.singletonList(new MemberIdentity())).build(requestHeader.apiVersion) - - when(controller.isActive).thenReturn(true) - - val request = TestUtils.buildEnvelopeRequest( - leaveGroupRequest, kafkaPrincipalSerde, requestChannelMetrics, time.nanoseconds()) - when(clientRequestQuotaManager.maybeRecordAndGetThrottleTimeMs(any[RequestChannel.Request](), - any[Long])).thenReturn(0) - kafkaApis = createKafkaApis(enableForwarding = true) - kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching) - - val response = verifyNoThrottling[EnvelopeResponse](request) - assertEquals(Errors.INVALID_REQUEST, response.error()) - } - - @Test - def testEnvelopeRequestWithNotFromPrivilegedListener(): Unit = { - testInvalidEnvelopeRequest(Errors.NONE, fromPrivilegedListener = false, - shouldCloseConnection = true) - } - - @Test - def testEnvelopeRequestNotAuthorized(): Unit = { - testInvalidEnvelopeRequest(Errors.CLUSTER_AUTHORIZATION_FAILED, - performAuthorize = true, authorizeResult = AuthorizationResult.DENIED) - } - - @Test - def testEnvelopeRequestNotControllerHandling(): Unit = { - testInvalidEnvelopeRequest(Errors.NOT_CONTROLLER, performAuthorize = true, isActiveController = false) - } - - private def testInvalidEnvelopeRequest(expectedError: Errors, - fromPrivilegedListener: Boolean = true, - shouldCloseConnection: Boolean = false, - performAuthorize: Boolean = false, - authorizeResult: AuthorizationResult = AuthorizationResult.ALLOWED, - isActiveController: Boolean = true): Unit = { - val authorizer: Authorizer = mock(classOf[Authorizer]) - - if (performAuthorize) { - authorizeResource(authorizer, AclOperation.CLUSTER_ACTION, ResourceType.CLUSTER, Resource.CLUSTER_NAME, authorizeResult) - } - - val resourceName = "topic-1" - val requestHeader = new RequestHeader(ApiKeys.ALTER_CONFIGS, ApiKeys.ALTER_CONFIGS.latestVersion, - clientId, 0) - - when(controller.isActive).thenReturn(isActiveController) - - val configResource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName) - - val configs = Map( - configResource -> new AlterConfigsRequest.Config( - Seq(new AlterConfigsRequest.ConfigEntry("foo", "bar")).asJava)) - val alterConfigsRequest = new AlterConfigsRequest.Builder(configs.asJava, false) - .build(requestHeader.apiVersion) - - val request = TestUtils.buildEnvelopeRequest( - alterConfigsRequest, kafkaPrincipalSerde, requestChannelMetrics, time.nanoseconds(), fromPrivilegedListener = fromPrivilegedListener) - - val capturedResponse: ArgumentCaptor[AbstractResponse] = ArgumentCaptor.forClass(classOf[AbstractResponse]) - kafkaApis = createKafkaApis(authorizer = Some(authorizer), enableForwarding = true) - kafkaApis.handle(request, RequestLocal.withThreadConfinedCaching) - - if (shouldCloseConnection) { - verify(requestChannel).closeConnection( - ArgumentMatchers.eq(request), - ArgumentMatchers.eq(java.util.Collections.emptyMap()) - ) - } else { - verify(requestChannel).sendResponse( - ArgumentMatchers.eq(request), - capturedResponse.capture(), - ArgumentMatchers.eq(None)) - val response = capturedResponse.getValue.asInstanceOf[EnvelopeResponse] - assertEquals(expectedError, response.error) - } - if (performAuthorize) { - verify(authorizer).authorize(any(), any()) - } - } - @Test def testAlterConfigsWithAuthorizer(): Unit = { val authorizer: Authorizer = mock(classOf[Authorizer]) @@ -10835,7 +10674,7 @@ class KafkaApisTest extends Logging { def testRaftShouldNeverHandleEnvelope(): Unit = { metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0) kafkaApis = createKafkaApis(raftSupport = true) - verifyShouldNeverHandleErrorMessage(kafkaApis.handleEnvelope(_, RequestLocal.withThreadConfinedCaching)) + verifyShouldNeverHandleErrorMessage(kafkaApis.handleEnvelope(_)) } @Test From 20c1a26321c15d85ea15147912c32941f8c3f8d4 Mon Sep 17 00:00:00 2001 From: TaiJu Wu Date: Thu, 9 Jan 2025 12:03:06 +0000 Subject: [PATCH 2/2] remove unnecessary call path --- core/src/main/scala/kafka/server/KafkaApis.scala | 10 ---------- .../scala/unit/kafka/server/KafkaApisTest.scala | 14 -------------- 2 files changed, 24 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 87423bb0d7f7a..32700b367082e 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -195,7 +195,6 @@ class KafkaApis(val requestChannel: RequestChannel, case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request) case ApiKeys.STOP_REPLICA => handleStopReplicaRequest(request) case ApiKeys.UPDATE_METADATA => handleUpdateMetadataRequest(request, requestLocal) - case ApiKeys.CONTROLLED_SHUTDOWN => handleControlledShutdownRequest(request) case ApiKeys.OFFSET_COMMIT => handleOffsetCommitRequest(request, requestLocal).exceptionally(handleError) case ApiKeys.OFFSET_FETCH => handleOffsetFetchRequest(request).exceptionally(handleError) case ApiKeys.FIND_COORDINATOR => handleFindCoordinatorRequest(request) @@ -244,7 +243,6 @@ class KafkaApis(val requestChannel: RequestChannel, case ApiKeys.ALTER_USER_SCRAM_CREDENTIALS => maybeForwardToController(request, handleAlterUserScramCredentialsRequest) case ApiKeys.ALTER_PARTITION => handleAlterPartitionRequest(request) case ApiKeys.UPDATE_FEATURES => maybeForwardToController(request, handleUpdateFeatures) - case ApiKeys.ENVELOPE => handleEnvelope(request) case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request) case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request) case ApiKeys.UNREGISTER_BROKER => forwardToControllerOrFail(request) @@ -419,10 +417,6 @@ class KafkaApis(val requestChannel: RequestChannel, } } - def handleControlledShutdownRequest(request: RequestChannel.Request): Unit = { - throw KafkaApis.shouldNeverReceive(request) - } - /** * Handle an offset commit request */ @@ -3460,10 +3454,6 @@ class KafkaApis(val requestChannel: RequestChannel, new DescribeClusterResponse(response.setThrottleTimeMs(requestThrottleMs))) } - def handleEnvelope(request: RequestChannel.Request): Unit = { - throw KafkaApis.shouldNeverReceive(request) - } - def handleDescribeProducersRequest(request: RequestChannel.Request): Unit = { val describeProducersRequest = request.body[DescribeProducersRequest] diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala index 387f168b8fa6f..55a728abdf71b 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala @@ -10656,13 +10656,6 @@ class KafkaApisTest extends Logging { verifyShouldNeverHandleErrorMessage(kafkaApis.handleUpdateMetadataRequest(_, RequestLocal.withThreadConfinedCaching)) } - @Test - def testRaftShouldNeverHandleControlledShutdownRequest(): Unit = { - metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0) - kafkaApis = createKafkaApis(raftSupport = true) - verifyShouldNeverHandleErrorMessage(kafkaApis.handleControlledShutdownRequest) - } - @Test def testRaftShouldNeverHandleAlterPartitionRequest(): Unit = { metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0) @@ -10670,13 +10663,6 @@ class KafkaApisTest extends Logging { verifyShouldNeverHandleErrorMessage(kafkaApis.handleAlterPartitionRequest) } - @Test - def testRaftShouldNeverHandleEnvelope(): Unit = { - metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0) - kafkaApis = createKafkaApis(raftSupport = true) - verifyShouldNeverHandleErrorMessage(kafkaApis.handleEnvelope(_)) - } - @Test def testRaftShouldAlwaysForwardCreateTopicsRequest(): Unit = { metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)