Skip to content
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

KAFKA-18399 Remove ZooKeeper from KafkaApis (5/N): ELECT_LEADERS , ALTER_PARTITION, UPDATE_FEATURES #18453

Open
wants to merge 4 commits into
base: trunk
Choose a base branch
from
+3 −110
Open

KAFKA-18399 Remove ZooKeeper from KafkaApis (5/N): ELECT_LEADERS , ALTER_PARTITION, UPDATE_FEATURES #18453

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ff10e4c
throw Exception
TaiJuWu Jan 8, 2025
bf28b40
remove ELECT_LEADERS ALTER_PARTITION UPDATE_FEATURES
TaiJuWu Jan 9, 2025
162d563
simpify change
TaiJuWu Jan 9, 2025
85f5a9b
Merge branch 'trunk' into rm_zk-4
TaiJuWu Jan 10, 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
129 changes: 14 additions & 115 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 @@ -43,7 +43,6 @@ import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic
import org.apache.kafka.common.message.CreateTopicsResponseData.{CreatableTopicResult, CreatableTopicResultCollection}
import org.apache.kafka.common.message.DeleteRecordsResponseData.{DeleteRecordsPartitionResult, DeleteRecordsTopicResult}
import org.apache.kafka.common.message.DeleteTopicsResponseData.{DeletableTopicResult, DeletableTopicResultCollection}
import org.apache.kafka.common.message.ElectLeadersResponseData.{PartitionResult, ReplicaElectionResult}
import org.apache.kafka.common.message.ListClientMetricsResourcesResponseData.ClientMetricsResource
import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition
import org.apache.kafka.common.message.ListOffsetsResponseData.{ListOffsetsPartitionResponse, ListOffsetsTopicResponse}
Expand All @@ -52,7 +51,7 @@ import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetFor
import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.{EpochEndOffset, OffsetForLeaderTopicResult, OffsetForLeaderTopicResultCollection}
import org.apache.kafka.common.message._
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.{ListenerName}
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.{ApiKeys, ApiMessage, Errors}
import org.apache.kafka.common.record._
import org.apache.kafka.common.replica.ClientMetadata
Expand Down Expand Up @@ -172,6 +171,16 @@ class KafkaApis(val requestChannel: RequestChannel,
maybeForwardToController(request, errorHandler)
}

private def forwardToControllerOrThrow(request: RequestChannel.Request,
createException: RequestChannel.Request => Exception
): Unit = {
def errorHandler(request: RequestChannel.Request): Unit = {
throw createException(request)
}

maybeForwardToController(request, errorHandler)
}

/**
* Top-level method that handles all requests and multiplexes to the right api
*/
Expand Down Expand Up @@ -238,7 +247,7 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.EXPIRE_DELEGATION_TOKEN => handleExpireTokenRequest(request)
case ApiKeys.DESCRIBE_DELEGATION_TOKEN => handleDescribeTokensRequest(request)
case ApiKeys.DELETE_GROUPS => handleDeleteGroupsRequest(request, requestLocal).exceptionally(handleError)
case ApiKeys.ELECT_LEADERS => maybeForwardToController(request, handleElectLeaders)
case ApiKeys.ELECT_LEADERS => forwardToControllerOrThrow(request, KafkaApis.shouldAlwaysForward)
TaiJuWu marked this conversation as resolved.
Show resolved Hide resolved
case ApiKeys.INCREMENTAL_ALTER_CONFIGS => handleIncrementalAlterConfigsRequest(request)
case ApiKeys.ALTER_PARTITION_REASSIGNMENTS => maybeForwardToController(request, handleAlterPartitionReassignmentsRequest)
case ApiKeys.LIST_PARTITION_REASSIGNMENTS => maybeForwardToController(request, handleListPartitionReassignmentsRequest)
Expand Down Expand Up @@ -3150,77 +3159,6 @@ class KafkaApis(val requestChannel: RequestChannel,
true
}

def handleElectLeaders(request: RequestChannel.Request): Unit = {
val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldAlwaysForward(request))
val electionRequest = request.body[ElectLeadersRequest]

def sendResponseCallback(
error: ApiError
)(
results: Map[TopicPartition, ApiError]
): Unit = {
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => {
val adjustedResults = if (electionRequest.data.topicPartitions == null) {
/* When performing elections across all of the partitions we should only return
* partitions for which there was an election or resulted in an error. In other
* words, partitions that didn't need election because they ready have the correct
* leader are not returned to the client.
*/
results.filter { case (_, error) =>
error.error != Errors.ELECTION_NOT_NEEDED
}
} else results

val electionResults = new util.ArrayList[ReplicaElectionResult]()
adjustedResults
.groupBy { case (tp, _) => tp.topic }
.foreachEntry { (topic, ps) =>
val electionResult = new ReplicaElectionResult()

electionResult.setTopic(topic)
ps.foreachEntry { (topicPartition, error) =>
val partitionResult = new PartitionResult()
partitionResult.setPartitionId(topicPartition.partition)
partitionResult.setErrorCode(error.error.code)
partitionResult.setErrorMessage(error.message)
electionResult.partitionResult.add(partitionResult)
}

electionResults.add(electionResult)
}

new ElectLeadersResponse(
requestThrottleMs,
error.error.code,
electionResults,
electionRequest.version
)
})
}

if (!authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME)) {
val error = new ApiError(Errors.CLUSTER_AUTHORIZATION_FAILED, null)
val partitionErrors: Map[TopicPartition, ApiError] =
electionRequest.topicPartitions.asScala.iterator.map(partition => partition -> error).toMap

sendResponseCallback(error)(partitionErrors)
} else {
val partitions = if (electionRequest.data.topicPartitions == null) {
metadataCache.getAllTopics().flatMap(metadataCache.getTopicPartitions)
} else {
electionRequest.topicPartitions.asScala
}

replicaManager.electLeaders(
zkSupport.controller,
partitions,
electionRequest.electionType,
sendResponseCallback(ApiError.NONE),
electionRequest.data.timeoutMs
)
}
}

def handleOffsetDeleteRequest(
request: RequestChannel.Request,
requestLocal: RequestLocal
Expand Down Expand Up @@ -3401,50 +3339,11 @@ class KafkaApis(val requestChannel: RequestChannel,
}

def handleAlterPartitionRequest(request: RequestChannel.Request): Unit = {
val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldNeverReceive(request))
val alterPartitionRequest = request.body[AlterPartitionRequest]
authHelper.authorizeClusterOperation(request, CLUSTER_ACTION)

if (!zkSupport.controller.isActive)
requestHelper.sendResponseExemptThrottle(request, alterPartitionRequest.getErrorResponse(
AbstractResponse.DEFAULT_THROTTLE_TIME, Errors.NOT_CONTROLLER.exception))
else
zkSupport.controller.alterPartitions(alterPartitionRequest.data, request.context.apiVersion, alterPartitionResp =>
requestHelper.sendResponseExemptThrottle(request, new AlterPartitionResponse(alterPartitionResp)))
throw KafkaApis.shouldNeverReceive(request)
}

def handleUpdateFeatures(request: RequestChannel.Request): Unit = {
val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldAlwaysForward(request))
val updateFeaturesRequest = request.body[UpdateFeaturesRequest]

def sendResponseCallback(errors: Either[ApiError, Map[String, ApiError]]): Unit = {
def createResponse(throttleTimeMs: Int): UpdateFeaturesResponse = {
errors match {
case Left(topLevelError) =>
UpdateFeaturesResponse.createWithErrors(
topLevelError,
Collections.emptySet(),
throttleTimeMs)
case Right(featureUpdateErrors) =>
// This response is not correct, but since this is ZK specific code it will be removed in 4.0
UpdateFeaturesResponse.createWithErrors(
ApiError.NONE,
featureUpdateErrors.asJava.keySet(),
throttleTimeMs)
}
}
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => createResponse(requestThrottleMs))
}

if (!authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME)) {
sendResponseCallback(Left(new ApiError(Errors.CLUSTER_AUTHORIZATION_FAILED)))
} else if (!zkSupport.controller.isActive) {
sendResponseCallback(Left(new ApiError(Errors.NOT_CONTROLLER)))
} else if (!config.isFeatureVersioningSupported) {
sendResponseCallback(Left(new ApiError(Errors.INVALID_REQUEST, "Feature versioning system is disabled.")))
} else {
zkSupport.controller.updateFeatures(updateFeaturesRequest, sendResponseCallback)
}
throw KafkaApis.shouldAlwaysForward(request)
}

def handleDescribeCluster(request: RequestChannel.Request): Unit = {
Expand Down
7 changes: 0 additions & 7 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 @@ -11004,13 +11004,6 @@ class KafkaApisTest extends Logging {
verifyShouldAlwaysForwardErrorMessage(kafkaApis.handleUpdateFeatures)
}

@Test
def testRaftShouldAlwaysForwardElectLeaders(): Unit = {
metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)
kafkaApis = createKafkaApis(raftSupport = true)
verifyShouldAlwaysForwardErrorMessage(kafkaApis.handleElectLeaders)
}

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