当前位置：首页 > news >正文

寻找协调器FindCoordinatorRequest请求流程

news 来源：原创 2024/5/19 20:54:10

文章目录

- 客户端发起请求
- Broker处理请求
- - 简单校验
  - 获取分区号和元信息
  - 构建返回数据 createResponse
- 问题

客户端发起请求

我们在分析消费者的时候, 有看到调用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.policy compact 日志清理策略为：紧缩
segment.bytes 10010241024 一个日志段的大小
compression.type producer 压缩类型为跟生产者保持一致

属性	值	描述
cleanup.policy	compact	日志清理策略为：紧缩
segment.bytes	10010241024	一个日志段的大小
compression.type	producer	压缩类型为跟生产者保持一致