Kafka 消费者之 findCoordinator源码解析

た入场券 2023-09-26 17:07 66阅读 0赞

### 1在这里我们先来梳理一下consumeGroup的相关知识 ###

1、首先，我们会给每个consume设置groupId，对于相同groupId且订阅相同topic的consume，会组成consumeGroup，如图一所示

![format_png][]

2、对于Server端的topic来说，会有partition这个概念，如图二所示

![format_png 1][]

图二

3、现在我们有多个consume及多个partition，到底由哪个consume来消费哪个partition呢？就由consume启动时的分区分配策略来决定。

*  如果consume数量小于partition的数量，则一个consume有可能消费多个分区，如图三所示
    
    ![format_png 2][]
 *  如果consume数量大于partition的数量，则会有consume线程空跑，如图四所示
    
    ![format_png 3][]
    
    4、kafka的内置topic：consumer\_offsets专门记录消费位点信息，既然是内置topic，那自然也会有partition及partition leader的概念，对于同一个groupId的消费位点都会记录在同一个partition中，在这篇文章中findCoordinator即是找到该groupId对应的partition的leader节点，我们知道这个节点才能将位移信息提交到这里保存，如果该partition还有其他副本，则该节点还会与其他副本同步位移信息。与该节点交互都是由GroupCoordinator完成的。

![format_png 4][]

在这里插入图片描述

### 2findCoordinator流程展示 ###

![format_png 5][]

在这里插入图片描述

### 3客户端源码分析 ###

这里还是放一下findCoordinator的代码，看其他consume的代码就发现客户端跟kafkaServer通信的格式大多是这样的，如果通信一次发现该GroupCoordinator的信息还未获取到则继续重试，直到超时，这里的超时时间即为poll时传入的超时时间，这个时间设置贯穿了整个consume的运行代码。

protected synchronized boolean ensureCoordinatorReady(final Timer timer) {
            //如果还未加入group则与group通信
            if (!coordinatorUnknown())
                return true;
    
            do {
                if (findCoordinatorException != null && !(findCoordinatorException instanceof RetriableException)) {
                    final RuntimeException fatalException = findCoordinatorException;
                    findCoordinatorException = null;
                    throw fatalException;
                }
                final RequestFuture<Void> future = lookupCoordinator();
                client.poll(future, timer);
              //如果还没回调完成则说明是超时的
                if (!future.isDone()) {
                    // ran out of time
                    break;
                }
    
                if (future.failed()) {
                    if (future.isRetriable()) {
                        log.debug("Coordinator discovery failed, refreshing metadata");
                        client.awaitMetadataUpdate(timer);
                    } else
                        throw future.exception();
                    //获取group的信息之后client会与group对应的节点建立连接，如果不可用则还会重试
                } else if (coordinator != null && client.isUnavailable(coordinator)) {
                    // we found the coordinator, but the connection has failed, so mark
                    // it dead and backoff before retrying discovery
                    markCoordinatorUnknown();
                    timer.sleep(rebalanceConfig.retryBackoffMs);
                }
                //如果与group通信成功则会跳出循环
            } while (coordinatorUnknown() && timer.notExpired());
    
            return !coordinatorUnknown();
        }
    复制代码

这里还有一点，跟踪代码可以看到以下代码在每次check以及与Server端通信完成之后都会有一样的逻辑，可以仔细思考一下，coordinator即获取到的group节点对象，client.isUnavailable(coordinator)是在与group建立连接，每次判断coordinator不为空且client与group连接失败，则将coordinator置空，为什么会这样呢？很有可能是请求到group的信息之后发现该节点已下线或者不可用，此时服务端很有可能也在进行选举，所以我们需要将coordinator清空，待服务端选举完成后再次通信。

protected synchronized Node checkAndGetCoordinator() {
            if (coordinator != null && client.isUnavailable(coordinator)) {
                markCoordinatorUnknown(true);
                return null;
            }
            return this.coordinator;
        }
    复制代码

#### org.apache.kafka.clients.consumer.internals.AbstractCoordinator\#lookupCoordinator ####

这段代码有个亮点就是先寻找的负载最小节点，然后与该节点通信获取group节点的信息。

protected synchronized RequestFuture<Void> lookupCoordinator() {
            if (findCoordinatorFuture == null) {
                // find a node to ask about the coordinator
                //与最小负载的node通信
                Node node = this.client.leastLoadedNode();
                if (node == null) {
                    log.debug("No broker available to send FindCoordinator request");
                    return RequestFuture.noBrokersAvailable();
                } else {
                    findCoordinatorFuture = sendFindCoordinatorRequest(node);
                    // remember the exception even after the future is cleared so that
                    // it can still be thrown by the ensureCoordinatorReady caller
                    findCoordinatorFuture.addListener(new RequestFutureListener<Void>() {
                        @Override
                        public void onSuccess(Void value) {} // do nothing
    
                        @Override
                        public void onFailure(RuntimeException e) {
                            findCoordinatorException = e;
                        }
                    });
                }
            }
            return findCoordinatorFuture;
        }
    复制代码

org.apache.kafka.clients.NetworkClient\#leastLoadedNode

我们先来看看是如何寻找负载最小节点的，这里代码还是挺讲究的，首先就是取随机数，防止每次都从第一个节点连接，如果判断没有在途的request则直接返回该节点，否则取在途request最小的节点，如果该节点不存在，则依次取连接的节点、需要重试的节点，如果找到不为null的节点则返回该节点，否则返回null。

public Node leastLoadedNode(long now) {
            List<Node> nodes = this.metadataUpdater.fetchNodes();
            if (nodes.isEmpty())
                throw new IllegalStateException("There are no nodes in the Kafka cluster");
            int inflight = Integer.MAX_VALUE;
    
            Node foundConnecting = null;
            Node foundCanConnect = null;
            Node foundReady = null;
            //随机取一个节点
            int offset = this.randOffset.nextInt(nodes.size());
            for (int i = 0; i < nodes.size(); i++) {
                int idx = (offset + i) % nodes.size();
                Node node = nodes.get(idx);
                //如果该节点是可连接的，且selector空闲，且发送队列空闲则可以发送请求
                if (canSendRequest(node.idString(), now)) {
                    //inFlightRequests记录了已发送请求但还未收到response的request，这里判定如果该节点没有这种数据则直接作为最小负载节点返回
                    int currInflight = this.inFlightRequests.count(node.idString());
                    if (currInflight == 0) {
                        // if we find an established connection with no in-flight requests we can stop right away
                        log.trace("Found least loaded node {} connected with no in-flight requests", node);
                        return node;
                        //否则取inFlightRequests中最小count的节点作为最小负载节点
                    } else if (currInflight < inflight) {
                        // otherwise if this is the best we have found so far, record that
                        inflight = currInflight;
                        foundReady = node;
                    }
                } else if (connectionStates.isPreparingConnection(node.idString())) {
                    foundConnecting = node;
                } else if (canConnect(node, now)) {
                    //如果该节点未被记录或者断连之后超过重试时间，则允许设置该节点
                    foundCanConnect = node;
                } else {
                    log.trace("Removing node {} from least loaded node selection since it is neither ready " +
                            "for sending or connecting", node);
                }
            }
    
            // We prefer established connections if possible. Otherwise, we will wait for connections
            // which are being established before connecting to new nodes.
            //优先取状态良好的节点
            if (foundReady != null) {
                log.trace("Found least loaded node {} with {} inflight requests", foundReady, inflight);
                return foundReady;
            } else if (foundConnecting != null) {
                log.trace("Found least loaded connecting node {}", foundConnecting);
                return foundConnecting;
            } else if (foundCanConnect != null) {
                log.trace("Found least loaded node {} with no active connection", foundCanConnect);
                return foundCanConnect;
            } else {
                log.trace("Least loaded node selection failed to find an available node");
                return null;
            }
        }
    复制代码

#### 拆解FindCoordinatorRequest ####

通过下图我们来看看发送了哪些数据，key\_type有两种枚举，一种是GROUP，另一种是TRANSACTION，如果type为GROUP的话那key就是groupId

![format_png 6][]

在这里插入图片描述

### 4服务端源码分析 ###

#### kafka.server.KafkaApis\#handleFindCoordinatorRequest ####

服务端还是通过KafkaApi来处理请求，代码也比较简单。

def handleFindCoordinatorRequest(request: RequestChannel.Request): Unit = {
        val findCoordinatorRequest = request.body[FindCoordinatorRequest]
        //校验数据
        //……省略部分代码
          // get metadata (and create the topic if necessary)
          val (partition, topicMetadata) = CoordinatorType.forId(findCoordinatorRequest.data.keyType) match {
            case CoordinatorType.GROUP =>
                //4.1 找到对应发分区
              val partition = groupCoordinator.partitionFor(findCoordinatorRequest.data.key)
                //4.2 获取对应的元数据
              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")
          }
          //组装返回参数
         //……省略部分代码
        }
      }
    复制代码

kafka.coordinator.group.GroupMetadataManager\#partitionFor

我们知道consume消费后对应的位点是保存在kafka的内部名为"\_\_consumer\_offsets"的内置topic中，内置topic初始化时由`offsets.topic.num.partitions`参数来决定分区数，默认值是50，相同consumeGroup的offset最终会保存在其中一个分区中，而保存在哪个分区就由下面这段代码来决定，可以看到逻辑很简单，就是取groupId的hashCode，然后对总的分区数取模。比如groupId为"consume\_group"，最终就会在34号分区保存位点。

def partitionFor(groupId: String): Int = Utils.abs(groupId.hashCode) % groupMetadataTopicPartitionCount
    复制代码

kafka.server.KafkaApis\#getOrCreateInternalTopic

这里是先从当前node的元数据缓存中拿到对应topic的数据，如果没有，则创建。从这段代码也可以猜想kafka内置topic的创建原理，是一种懒加载的思想，当第一个consume接入之后才会创建对应topicPartition文件。

private def getOrCreateInternalTopic(topic: String, listenerName: ListenerName): MetadataResponse.TopicMetadata = {
        val topicMetadata = metadataCache.getTopicMetadata(Set(topic), listenerName)
        topicMetadata.headOption.getOrElse(createInternalTopic(topic))
      }
    复制代码

这里的topicMetadata就是对应入参topic返回类似列表的对象，因为入参只有一个topic，所以直接取第一个数据，数据结构见下图，可以更直观的理解返回参数。

![format_png 7][]

在这里插入图片描述

![format_png 8][]

在这里插入图片描述

kafka.server.KafkaApis\#createTopic

Topic创建的流程如下图所示，详情请看 [Topic创建流程源码分析][Topic]

![format_png 9][]

#### 拆解FindCoordinatorResponse ####

通过下图我们来看看返回了哪些数据，可以看到前面取了很多数据，最终拼到返回参数里面的只有leader所在的节点信息

![format_png 10][]

在这里插入图片描述

### 5总结 ###

这块代码本身不是很复杂，主要是有一些细节需要考虑，通过仔细思量这些细节对我们今后分析consume异常会大有好处。流程总结如下 1、寻找最小负载节点信息 2、向最小负载节点发送FindCoordinatorRequest 3、最小负载节点处理该请求。

*  首先找到该groupId对应的分区
 *  通过内存中缓存的metaData获取该分区的信息，如果不存在则创建topic
 *  返回查找到的分区leader信息

[format_png]: https://img-blog.csdnimg.cn/img_convert/6fd80c7f2a49d3308cdab71f78aca46c.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/30ad9fe046677cd2e6794ca95f36f8db.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/74e7a90e2189a55a399165ee0546263b.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/2fa6a0558844c834c986d74f7a5af36b.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/15c2c85ca40fe80321f5d902361361b9.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/a88cf734c944a270b12cef7b22f41338.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/305b69a851bf2e84a16e49cd13399422.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/ec6cc9c0582a2671aeaca681717d3532.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/5da6458246226856bc027b3a95668d23.webp?x-oss-process=image/format,png
[Topic]: https://link.juejin.cn?target=https%3A%2F%2Fmp.weixin.qq.com%2Fs%3F__biz%3DMzg4ODY1NTcxNg%3D%3D%26mid%3D2247485101%26idx%3D1%26sn%3De89ee4792d3798a2f6ca5f5ad9b69464
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/106b554338b5de5fd57102a761eda9a9.webp?x-oss-process=image/format,png
[format_png 10]: https://img-blog.csdnimg.cn/img_convert/57380d3689d08826f137cb4f66ee6c9c.webp?x-oss-process=image/format,png