寻找协调器FindCoordinatorRequest请求流程

石臻臻的杂货铺[同名公众号]

发布于 2022-09-28 04:57:36

30800

代码可运行

文章被收录于专栏：kafka专栏kafka专栏

运行总次数：0

代码可运行

文章目录

- [客户端发起请求](https://cloud.tencent.com/developer)
- [Broker处理请求](https://cloud.tencent.com/developer)
    - [简单校验](https://cloud.tencent.com/developer)
    - [获取分区号和元信息](https://cloud.tencent.com/developer)
    - [构建返回数据 createResponse](https://cloud.tencent.com/developer)
- [问题](https://cloud.tencent.com/developer)

客户端发起请求

我们在分析消费者的时候, 有看到调用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());
    }

Broker处理请求

def handleFindCoordinatorRequest(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)
    else if (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 {
        case CoordinatorType.GROUP =>
          val partition = groupCoordinator.partitionFor(findCoordinatorRequest.data.key)
          val metadata = getOrCreateInternalTopic(GROUP_METADATA_TOPIC_NAME, request.context.listenerName)
          (partition, metadata)

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

        case _ =>
          throw new InvalidRequestException("Unknown coordinator type in FindCoordinator request")
      }

      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 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)
          }
        }
        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的情况

去zk获取/brokers/topic/__consumer_offsets 数据找到消费者Topic的分区总数。默认是50. （由offsets.topic.num.partitions 控制）找到分区数之和后, 则计算 Utils.abs(groupId.hashCode) % groupMetadataTopicPartitionCount(groupID按分区数取模运算)获取到了分区号partition;
然后接着获取该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 协议。
如果发现没有这个Topic的元信息，则需要去创建__consumer_offsetsTopic 。注意：创建这个Topic的的几个特殊属性: 属性值描述cleanup.policycompact日志清理策略为：紧缩segment.bytes100_1024_1024一个日志段的大小compression.typeproducer压缩类型为跟生产者保持一致