寻找协调器FindCoordinatorRequest请求流程

kafka管控推荐使用滴滴开源的 Kafka运维管控平台更符合国人的操作习惯 ,
更强大的管控能力 ,更高效的问题定位能力、更便捷的集群运维能力、更专业的资源治理、更友好的运维生态

客户端发起请求

我们在分析消费者的时候, 有看到调用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	压缩类型为跟生产者保持一致