寻找协调器FindCoordinatorRequest请求流程

提示：本文可能已过期,请点击原文查看：寻找协调器FindCoordinatorRequest请求流程

@

• 客户端发起请求
• Broker处理请求

• 简单校验
• 获取分区号和元信息
• 构建返回数据 createResponse

• 问题

1客户端发起请求

我们在分析消费者的时候, 有看到调用FindCoordinatorRequest的请求


   private RequestFuture<Void> sendFindCoordinatorRequest(Node node) {
       // initiate the group metadata request
       log.debug("Sending FindCoordinator request to broker {}", node);
       FindCoordinatorRequest.Builder requestBuilder =
               new FindCoordinatorRequest.Builder(
                       new FindCoordinatorRequestData()
                           .setKeyType(CoordinatorType.GROUP.id())
                           .setKey(this.rebalanceConfig.groupId));
       return client.send(node, requestBuilder)
               .compose(new FindCoordinatorResponseHandler());
   }
   


2Broker处理请求



defhandleFindCoordinatorRequest(request: RequestChannel.Request): Unit = {
   val findCoordinatorRequest = request.body[FindCoordinatorRequest]

// 根据协调器类型判断是否授权过
   if (findCoordinatorRequest.data.keyType == CoordinatorType.GROUP.id &&
       !authorize(request.context, DESCRIBE, GROUP, findCoordinatorRequest.data.key))
     sendErrorResponseMaybeThrottle(request, Errors.GROUP_AUTHORIZATION_FAILED.exception)
   elseif (findCoordinatorRequest.data.keyType == CoordinatorType.TRANSACTION.id &&
       !authorize(request.context, DESCRIBE, TRANSACTIONAL_ID, findCoordinatorRequest.data.key))
     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 {
       caseCoordinatorType.GROUP =>
         val partition = groupCoordinator.partitionFor(findCoordinatorRequest.data.key)
         val metadata = getOrCreateInternalTopic(GROUP_METADATA_TOPIC_NAME, request.context.listenerName)
         (partition, metadata)

       caseCoordinatorType.TRANSACTION =>
         val partition = txnCoordinator.partitionFor(findCoordinatorRequest.data.key)
         val metadata = getOrCreateInternalTopic(TRANSACTION_STATE_TOPIC_NAME, request.context.listenerName)
         (partition, metadata)

       case _ =>
         thrownewInvalidRequestException("Unknown coordinator type in FindCoordinator request")
     }

     defcreateResponse(requestThrottleMs: Int): AbstractResponse = {
       defcreateFindCoordinatorResponse(error: Errors, node: Node): FindCoordinatorResponse = {
         newFindCoordinatorResponse(
             newFindCoordinatorResponseData()
               .setErrorCode(error.code)
               .setErrorMessage(error.message)
               .setNodeId(node.id)
               .setHost(node.host)
               .setPort(node.port)
               .setThrottleTimeMs(requestThrottleMs))
       }
       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 {
           caseSome(endpoint) =>
             createFindCoordinatorResponse(Errors.NONE, endpoint)
           case _ =>
             createFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode)
         }
       }
       trace("Sending FindCoordinator response %s for correlation id %d to client %s."
         .format(responseBody, request.header.correlationId, request.header.clientId))
       responseBody
     }
     sendResponseMaybeThrottle(request, createResponse)
   }
 }





简单校验

根据协调器类型判断是否有被授权。协调器类型有  GROUP((byte) 0), TRANSACTION((byte) 1)两种

获取分区号和元信息

这里的接口分两种情况，一个是协调列席为GROUP  一个是 TRANSACTION他们的处理逻辑都是一样的,只是处理的Topic不一样

GROUP 对应的Topic是  __consumer_offsets

TRANSACTION 对应的Topic是__transaction_state

这里我们主要分析一下 GROUP的情况

1. 去zk获取/brokers/topic/__consumer_offsets 数据  找到消费者Topic的分区总数。默认是50.  （由offsets.topic.num.partitions 控制）找到分区数之和后, 则计算 Utils.abs(groupId.hashCode) % groupMetadataTopicPartitionCount(groupID按分区数取模运算)获取到了分区号partition;
   
2. 然后接着获取该Topic的元信息, 这里需要注意的是 去获取元信息应该走的是什么 监听协议(listenerName) 呢？这个主要是看当前处理请求的Broker是通过哪个入口来的。比如说该Broker有两个监听口,listeners = INTER://xxx.xx.xx.100:9091, OUTSIDE://xxx.xx.xx.101:9092 .如果客户端发起请求的时候是对xxx.xx.xx.101:9092发起的请求，那么这个对应的监听器就是 OUTSIDE .  那么Broker去获取__consumer_offsets元信息发起请求的时候也是会用的 OUTSIDE 协议。
   
3. 如果发现没有这个Topic的元信息，则需要去创建__consumer_offsetsTopic 。 注意：创建这个Topic的的几个特殊属性:

构建返回数据 createResponse

这里才是真正的找到协调器的主要逻辑, 这里的判断逻辑是

上面我们获取到的分区号是partition， 我们同样获取到了__consumer_offsets的元信息Metadata。

那我们就可以获取到这个分区号, 并且就能够找到该分区的LeaderId所属在哪个Broker上。

知道了哪个Broker, 那我们就能够获取到对应的EndPoint, 一个Broker可能同时有多个EndPoint（配置了多个监听器），那么我们应该使用哪个EndPoint呢？

这个的判断逻辑与上面说过的一样，客户端发起请求时候的监听器是哪个，那么这里就应该用哪个监听器。

在这里插入图片描述

注意：如果找到的分区Leader不存在 那么这个协调器就不存在

然后会返回异常：


The coordinator is not available



3问题

1. 如果客户端走的外网监听器访问的集群，那么在客户端发起请求之后到集群内部，触发内部调用链的请求，那么内部这个调用链是用什么监听器访问的呢？

从客户端 -> Broker -> 其他Broker. 这是一个调用链路，从最开始用的是什么监听器那么这条链路上都是用的这个监听器！具体请看：多网络情况下,Kafka客户端如何选择合适的网络发起请求