Skip to content

KAFKA-18399 Remove ZooKeeper from KafkaApis (2/N): CONTROLLED_SHUTDOWN , ENVELOPE #18422

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 3 commits into from
Jan 9, 2025
Merged

KAFKA-18399 Remove ZooKeeper from KafkaApis (2/N): CONTROLLED_SHUTDOWN , ENVELOPE #18422

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f667edc
remove envelop
TaiJuWu Jan 8, 2025
20c1a26
remove unnecessary call path
TaiJuWu Jan 9, 2025
c2154b6
Merge branch 'trunk' into rm_zk-1
TaiJuWu Jan 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 0 additions & 57 deletions core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,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
Expand Down Expand Up @@ -137,10 +136,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
Expand Down Expand Up @@ -196,7 +191,6 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.FETCH => handleFetchRequest(request)
case ApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)
case ApiKeys.METADATA => handleTopicMetadataRequest(request)
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)
Expand Down Expand Up @@ -245,7 +239,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, requestLocal)

@chia7712 chia7712 Jan 8, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems to me using KafkaApis.shouldNeverReceive(request) is more readable.

@TaiJuWu TaiJuWu Jan 8, 2025

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. Thanks for review.

case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request)
case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request)
case ApiKeys.UNREGISTER_BROKER => forwardToControllerOrFail(request)
Expand Down Expand Up @@ -291,27 +284,6 @@ class KafkaApis(val requestChannel: RequestChannel,
replicaManager.tryCompleteActions()
}

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)
}

/**
* Handle an offset commit request
*/
Expand Down Expand Up @@ -3349,35 +3321,6 @@ 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 handleDescribeProducersRequest(request: RequestChannel.Request): Unit = {
val describeProducersRequest = request.body[DescribeProducersRequest]

Expand Down
175 changes: 0 additions & 175 deletions core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,167 +322,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])
Expand Down Expand Up @@ -10516,27 +10355,13 @@ class KafkaApisTest extends Logging {
assertEquals(KafkaApis.shouldAlwaysForward(request).getMessage, e.getMessage)
}

@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)
kafkaApis = createKafkaApis(raftSupport = true)
verifyShouldNeverHandleErrorMessage(kafkaApis.handleAlterPartitionRequest)
}

@Test
def testRaftShouldNeverHandleEnvelope(): Unit = {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)
kafkaApis = createKafkaApis(raftSupport = true)
verifyShouldNeverHandleErrorMessage(kafkaApis.handleEnvelope(_, RequestLocal.withThreadConfinedCaching))
}

@Test
def testRaftShouldAlwaysForwardCreateTopicsRequest(): Unit = {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)
Expand Down