abbccdda commented on a change in pull request #9579:
URL: https://github.com/apache/kafka/pull/9579#discussion_r561594743
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -1370,55 +1345,164 @@ class KafkaApis(val requestChannel: RequestChannel,
!authHelper.authorize(request.context, DESCRIBE, TRANSACTIONAL_ID,
findCoordinatorRequest.data.key))
requestHelper.sendErrorResponseMaybeThrottle(request,
Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED.exception)
else {
- // get metadata (and create the topic if necessary)
- val (partition, topicMetadata) =
CoordinatorType.forId(findCoordinatorRequest.data.keyType) match {
+ val (partition, internalTopicName) =
CoordinatorType.forId(findCoordinatorRequest.data.keyType) match {
case CoordinatorType.GROUP =>
- val partition =
groupCoordinator.partitionFor(findCoordinatorRequest.data.key)
- val metadata = getOrCreateInternalTopic(GROUP_METADATA_TOPIC_NAME,
request.context.listenerName)
- (partition, metadata)
+ (groupCoordinator.partitionFor(findCoordinatorRequest.data.key),
GROUP_METADATA_TOPIC_NAME)
case CoordinatorType.TRANSACTION =>
- val partition =
txnCoordinator.partitionFor(findCoordinatorRequest.data.key)
- val metadata =
getOrCreateInternalTopic(TRANSACTION_STATE_TOPIC_NAME,
request.context.listenerName)
- (partition, metadata)
+ (txnCoordinator.partitionFor(findCoordinatorRequest.data.key),
TRANSACTION_STATE_TOPIC_NAME)
+ }
- case _ =>
- throw new InvalidRequestException("Unknown coordinator type in
FindCoordinator request")
+ val topicMetadata =
metadataCache.getTopicMetadata(Set(internalTopicName),
request.context.listenerName)
+ def createFindCoordinatorResponse(error: Errors,
+ node: Node,
+ requestThrottleMs: Int,
+ errorMessage: Option[String] = None):
FindCoordinatorResponse = {
+ new FindCoordinatorResponse(
+ new FindCoordinatorResponseData()
+ .setErrorCode(error.code)
+ .setErrorMessage(errorMessage.getOrElse(error.message))
+ .setNodeId(node.id)
+ .setHost(node.host)
+ .setPort(node.port)
+ .setThrottleTimeMs(requestThrottleMs))
}
- def createResponse(requestThrottleMs: Int): AbstractResponse = {
- def createFindCoordinatorResponse(error: Errors, node: Node):
FindCoordinatorResponse = {
- new FindCoordinatorResponse(
- new FindCoordinatorResponseData()
- .setErrorCode(error.code)
- .setErrorMessage(error.message)
- .setNodeId(node.id)
- .setHost(node.host)
- .setPort(node.port)
- .setThrottleTimeMs(requestThrottleMs))
+ val topicCreationNeeded = topicMetadata.headOption.isEmpty
+ if (topicCreationNeeded) {
+ if (hasEnoughAliveBrokers(internalTopicName)) {
+ if (shouldForwardRequest(request)) {
+ forwardingManager.sendInterBrokerRequest(
+ getCreateTopicsRequest(Seq(internalTopicName)),
+ _ => ())
+ } else {
+ val controllerMutationQuota =
quotas.controllerMutation.newQuotaFor(request, strictSinceVersion = 6)
+
+ val topicConfigs = Map(internalTopicName ->
getTopicConfigs(internalTopicName))
+ adminManager.createTopics(
+ config.requestTimeoutMs,
+ validateOnly = false,
+ topicConfigs,
+ Map.empty,
+ controllerMutationQuota,
+ _ => ())
+ }
}
- val responseBody = if (topicMetadata.errorCode != Errors.NONE.code) {
- createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE,
Node.noNode)
- } else {
- val coordinatorEndpoint = topicMetadata.partitions.asScala
- .find(_.partitionIndex == partition)
- .filter(_.leaderId != MetadataResponse.NO_LEADER_ID)
- .flatMap(metadata =>
metadataCache.getAliveBroker(metadata.leaderId))
- .flatMap(_.getNode(request.context.listenerName))
- .filterNot(_.isEmpty)
-
- coordinatorEndpoint match {
- case Some(endpoint) =>
- createFindCoordinatorResponse(Errors.NONE, endpoint)
- case _ =>
- createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE,
Node.noNode)
+
+ requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
createFindCoordinatorResponse(
+ Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode, requestThrottleMs))
+ } else {
+ def createResponse(requestThrottleMs: Int): AbstractResponse = {
+ val responseBody = if (topicMetadata.head.errorCode !=
Errors.NONE.code) {
+ createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE,
Node.noNode, requestThrottleMs)
+ } else {
+ val coordinatorEndpoint = topicMetadata.head.partitions.asScala
+ .find(_.partitionIndex == partition)
+ .filter(_.leaderId != MetadataResponse.NO_LEADER_ID)
+ .flatMap(metadata =>
metadataCache.getAliveBroker(metadata.leaderId))
+ .flatMap(_.getNode(request.context.listenerName))
+ .filterNot(_.isEmpty)
+
+ coordinatorEndpoint match {
+ case Some(endpoint) =>
+ createFindCoordinatorResponse(Errors.NONE, endpoint,
requestThrottleMs)
+ case _ =>
+
createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode,
requestThrottleMs)
+ }
}
+ trace("Sending FindCoordinator response %s for correlation id %d to
client %s."
+ .format(responseBody, request.header.correlationId,
request.header.clientId))
+ responseBody
}
- trace("Sending FindCoordinator response %s for correlation id %d to
client %s."
- .format(responseBody, request.header.correlationId,
request.header.clientId))
- responseBody
+
+ requestHelper.sendResponseMaybeThrottle(request, createResponse)
+ }
+ }
+ }
+
+ private def getCreateTopicsRequest(topics: Seq[String]):
CreateTopicsRequest.Builder = {
+ val topicCollection = new CreateTopicsRequestData.CreatableTopicCollection
+ topics.foreach(topic => {
+ topicCollection.add(getTopicConfigs(topic))
+ })
+
+ new CreateTopicsRequest.Builder(
+ new CreateTopicsRequestData()
+ .setTimeoutMs(config.requestTimeoutMs)
+ .setTopics(topicCollection)
+ )
+ }
+
+ private def getTopicConfigs(topic: String): CreatableTopic = {
+ topic match {
+ case GROUP_METADATA_TOPIC_NAME =>
+ new CreatableTopic()
+ .setName(topic)
+ .setNumPartitions(config.offsetsTopicPartitions)
Review comment:
I agree it's equivalent, but I think we could be conservative here to
keep the logic on broker side for now, to reduce logical change in this PR.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
