Kafka消费者JoinGroupRequest流程解析

RequestFuture<ByteBuffer> sendJoinGroupRequest() {
  // 如果 协调器未获取到或者链接断开了 则返回异常
        if (coordinatorUnknown())
            return RequestFuture.coordinatorNotAvailable();
  // 构建请求参数
        JoinGroupRequest.Builder requestBuilder = new JoinGroupRequest.Builder(
                new JoinGroupRequestData()
                        .setGroupId(rebalanceConfig.groupId)
                        // 客户端与Broker最大会话有效期, 如果超过这个时间没有任何心跳则可能发生重平衡
                        // 属性 session.timeout.ms；默认10000（10 秒）
                        .setSessionTimeoutMs(this.rebalanceConfig.sessionTimeoutMs)
                        //消费者的成员id， 默认就是空字符串 
                        .setMemberId(this.generation.memberId)
                        //2.3版本引入的新参数. 用户提供的消费者实例的唯一标识符。 如果设置了,消费者则被视为静态成员, 
                        //静态成员配以较大的session超时设置能够避免因成员临时不可用（比如重启）而引发的Rebalance，
                        // 如果不设置, 消费者将作为动态成员
                        //配置：group.instance.id  
                        .setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null))
                        //协议类型, 这里是consumer, 可选项还有connect 
                        .setProtocolType(protocolType())
                        // 配置的分区分配策略和对应的 订阅相关信息的
                        .setProtocols(metadata())
                        //
                        .setRebalanceTimeoutMs(this.rebalanceConfig.rebalanceTimeoutMs)
        );
        log.debug("Sending JoinGroup ({}) to coordinator {}", requestBuilder, this.coordinator);
        // joinGroup请求的超时时间 在 (request.timeout.ms默认30000（30 秒）) 和(max.poll.interval.ms默认30s + 5s ） 中取最大值
        int joinGroupTimeoutMs = Math.max(client.defaultRequestTimeoutMs(),
            rebalanceConfig.rebalanceTimeoutMs + JOIN_GROUP_TIMEOUT_LAPSE);
        // 向指定的coordinator  Node节点    发起请求
        return client.send(coordinator, requestBuilder, joinGroupTimeoutMs)
                .compose(new JoinGroupResponseHandler(generation));
    }

public static final Schema[] SCHEMAS = new Schema[] {
        SCHEMA_0,
        SCHEMA_1,
        SCHEMA_2,
        SCHEMA_3,
        SCHEMA_4,
        SCHEMA_5,
        SCHEMA_6,
        SCHEMA_7
    };

@Override
    protected JoinGroupRequestData.JoinGroupRequestProtocolCollection metadata() {
        log.debug("Joining group with current subscription: {}", subscriptions.subscription());
        this.joinedSubscription = subscriptions.subscription();
        JoinGroupRequestData.JoinGroupRequestProtocolCollection protocolSet = new JoinGroupRequestData.JoinGroupRequestProtocolCollection();
        List<String> topics = new ArrayList<>(joinedSubscription);
        // assignors 是配置 partition.assignment.strategy 的分区分配策略
        for (ConsumerPartitionAssignor assignor : assignors) {
            Subscription subscription = new Subscription(topics,
          // 分配策略的用户自定义数据,需要实现subscriptionUserData方法,默认range策略是null
                                         assignor.subscriptionUserData(joinedSubscription),
                                         // 订阅的所有分区列表
                                         subscriptions.assignedPartitionsList());
            //这个数据包含了 订阅的Topic、UserData 和经过分组整理的订阅的(Topic-分区列表  )  并且序列化成ByteBuffer                          
            ByteBuffer metadata = ConsumerProtocol.serializeSubscription(subscription);
   // 组装成 每种策略对应的 订阅信息数据
            protocolSet.add(new JoinGroupRequestData.JoinGroupRequestProtocol()
                    .setName(assignor.name())
                    .setMetadata(Utils.toArray(metadata)));
        }
        return protocolSet;
    }

client.send(coordinator, requestBuilder, joinGroupTimeoutMs)
                .compose(new JoinGroupResponseHandler(generation));

private class FindCoordinatorResponseHandler extends RequestFutureAdapter<ClientResponse, Void> {
        @Override
        public void onSuccess(ClientResponse resp, RequestFuture<Void> future) {
            log.debug("Received FindCoordinator response {}", resp);
            FindCoordinatorResponse findCoordinatorResponse = (FindCoordinatorResponse) resp.responseBody();
            Errors error = findCoordinatorResponse.error();
            if (error == Errors.NONE) {
                synchronized (AbstractCoordinator.this) {
                    // use MAX_VALUE - node.id as the coordinator id to allow separate connections
                    // for the coordinator in the underlying network client layer
                    int coordinatorConnectionId = Integer.MAX_VALUE - findCoordinatorResponse.data().nodeId();
     // 获取到了节点
                    AbstractCoordinator.this.coordinator = new Node(
                            coordinatorConnectionId,
                            findCoordinatorResponse.data().host(),
                            findCoordinatorResponse.data().port());
                    log.info("Discovered group coordinator {}", coordinator);
                    client.tryConnect(coordinator);
                    heartbeat.resetSessionTimeout();
                }
                future.complete(null);
            } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
                future.raise(GroupAuthorizationException.forGroupId(rebalanceConfig.groupId));
            } else {
                log.debug("Group coordinator lookup failed: {}", findCoordinatorResponse.data().errorMessage());
                future.raise(error);
            }
        }

//省略

The session timeout is not within the range allowed by the broker as configured by
       group.min.session.timeout.ms and group.max.session.timeout.ms)

The consumer group has reached its max size

private def doUnknownJoinGroup(group: GroupMetadata,
                                 groupInstanceId: Option[String],
                                 requireKnownMemberId: Boolean,
                                 clientId: String,
                                 clientHost: String,
                                 rebalanceTimeoutMs: Int,
                                 sessionTimeoutMs: Int,
                                 protocolType: String,
                                 protocols: List[(String, Array[Byte])],
                                 responseCallback: JoinCallback): Unit = {
    group.inLock {
      if (group.is(Dead)) {
            // 如果该组被标记为死亡，则意味着其他某个线程刚刚从协调器元数据中删除了该组；
        // 该组很可能已迁移到其他一些协调器，或者该组处于瞬态不稳定阶段。让成员重试找到正确的协调者并重新加入。
        responseCallback(JoinGroupResult(JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.COORDINATOR_NOT_AVAILABLE))
      // 检查group支持的协议
      } else if (!group.supportsProtocols(protocolType, MemberMetadata.plainProtocolSet(protocols))) {
        responseCallback(JoinGroupResult(JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.INCONSISTENT_GROUP_PROTOCOL))
      } else {
        //newMemberId计算逻辑 无groupInstanceId情况 格式： clientId-UUID ； 
        //有groupInstanceId情况 格式： groupInstanceId-UUID
        val newMemberId = group.generateMemberId(clientId, groupInstanceId)
        if (group.hasStaticMember(groupInstanceId)) {
          // 静态member的处理情况
          updateStaticMemberAndRebalance(group, newMemberId, groupInstanceId, protocols, responseCallback)
        } else if (requireKnownMemberId) {
            // 返回MEMBER_ID_REQUIRED异常,让客户端再发一次请求
          debug(s"Dynamic member with unknown member id joins group ${group.groupId} in " +
              s"${group.currentState} state. Created a new member id $newMemberId and request the member to rejoin with this id.")
          group.addPendingMember(newMemberId)
          addPendingMemberExpiration(group, newMemberId, sessionTimeoutMs)
          responseCallback(JoinGroupResult(newMemberId, Errors.MEMBER_ID_REQUIRED))
        } else {
          info(s"${if (groupInstanceId.isDefined) "Static" else "Dynamic"} Member with unknown member id joins group ${group.groupId} in " +
            s"${group.currentState} state. Created a new member id $newMemberId for this member and add to the group.")
          addMemberAndRebalance(rebalanceTimeoutMs, sessionTimeoutMs, newMemberId, groupInstanceId,
            clientId, clientHost, protocolType, protocols, group, responseCallback)
        }
      }
    }
  }

private def doJoinGroup(group: GroupMetadata,
                          memberId: String,
                          groupInstanceId: Option[String],
                          clientId: String,
                          clientHost: String,
                          rebalanceTimeoutMs: Int,
                          sessionTimeoutMs: Int,
                          protocolType: String,
                          protocols: List[(String, Array[Byte])],
                          responseCallback: JoinCallback): Unit = {
    group.inLock {
      if (group.is(Dead)) {
        // 如果该组被标记为死亡，则意味着其他某个线程刚刚从协调器元数据中删除了该组；
        // 这可能是该组已迁移到其他一些协调器，或者该组处于瞬态不稳定阶段。让成员重试找到正确的协调者并重新加入
        responseCallback(JoinGroupResult(memberId, Errors.COORDINATOR_NOT_AVAILABLE))
      } else if (!group.supportsProtocols(protocolType, MemberMetadata.plainProtocolSet(protocols))) {
        responseCallback(JoinGroupResult(memberId, Errors.INCONSISTENT_GROUP_PROTOCOL))
      } else if (group.isPendingMember(memberId)) {
        // 重新加入的待定成员将被接受。请注意，待定成员永远不会是静态成员。
        if (groupInstanceId.isDefined) {
          throw new IllegalStateException(s"the static member $groupInstanceId was not expected to be assigned " +
            s"into pending member bucket with member id $memberId")
        } else {
          debug(s"Dynamic Member with specific member id $memberId joins group ${group.groupId} in " +
            s"${group.currentState} state. Adding to the group now.")
          addMemberAndRebalance(rebalanceTimeoutMs, sessionTimeoutMs, memberId, groupInstanceId,
            clientId, clientHost, protocolType, protocols, group, responseCallback)
        }
      } else {
       // 这里的逻辑 省略...
    }
  }
  private def addMemberAndRebalance(rebalanceTimeoutMs: Int,
                                    sessionTimeoutMs: Int,
                                    memberId: String,
                                    groupInstanceId: Option[String],
                                    clientId: String,
                                    clientHost: String,
                                    protocolType: String,
                                    protocols: List[(String, Array[Byte])],
                                    group: GroupMetadata,
                                    callback: JoinCallback): Unit = {
    // 构建member对象
    val member = new MemberMetadata(memberId, group.groupId, groupInstanceId,
      clientId, clientHost, rebalanceTimeoutMs,
      sessionTimeoutMs, protocolType, protocols)
    member.isNew = true
    // update the newMemberAdded flag to indicate that the join group can be further delayed
    if (group.is(PreparingRebalance) && group.generationId == 0)
      group.newMemberAdded = true
  // 添加member，如果有callback！=null的话，则numMembersAwaitingJoin+1.
  //意思是又有一个member正在等待加入。
    group.add(member, callback)
    // 重新设置一下心跳检测,并把设置设置为了5分钟, 一开始的时候是10s,意味着10s内要发起第二次的JoinGroup才行
    // 但是走到了这里,把过期时间设置更大了一点, 这个时候客户端是还没有启动心跳线程的。
    completeAndScheduleNextExpiration(group, member, NewMemberJoinTimeoutMs)
    if (member.isStaticMember) {
      info(s"Adding new static member $groupInstanceId to group ${group.groupId} with member id $memberId.")
      group.addStaticMember(groupInstanceId, memberId)
    } else {
      // 把memberId从Pending缓存中移除, 因为已经在Group总添加了Member了
      group.removePendingMember(memberId)
    }
    //尝试将group状态流转为  PreparingRebalance，并且发起再平衡
    maybePrepareRebalance(group, s"Adding new member $memberId with group instance id $groupInstanceId")
  }

def onCompleteJoin(group: GroupMetadata): Unit = {
    group.inLock {
      val notYetRejoinedDynamicMembers = group.notYetRejoinedMembers.filterNot(_._2.isStaticMember)
      if (notYetRejoinedDynamicMembers.nonEmpty) {
        info(s"Group ${group.groupId} remove dynamic members " +
          s"who haven't joined: ${notYetRejoinedDynamicMembers.keySet}")
          // 还有一些Member没有重新Join的 直接在Group中把他们给移除掉,心跳监测也移除掉
          // 是否重新Rejoin的判断逻辑是, 请求的callback一只没有执行和设置为null。
        notYetRejoinedDynamicMembers.values foreach { failedMember =>
          //移除那些被移除的member的心跳监测  并在group中移除member
          removeHeartbeatForLeavingMember(group, failedMember)
          group.remove(failedMember.memberId)
          // TODO: cut the socket connection to the client
        }
      }
      if (group.is(Dead)) {
        info(s"Group ${group.groupId} is dead, skipping rebalance stage")
      } else if (!group.maybeElectNewJoinedLeader() && group.allMembers.nonEmpty) {
        //如果所有成员都没有重新加入，我们将推迟重新平衡准备阶段的完成，并发出另一个延迟操作，直到会话超时删除所有未响应的成员。
        error(s"Group ${group.groupId} could not complete rebalance because no members rejoined")
        joinPurgatory.tryCompleteElseWatch(
          new DelayedJoin(this, group, group.rebalanceTimeoutMs),
          Seq(GroupKey(group.groupId)))
      } else {
           // 初始化下个Generation对象，Generation自增，如果member是空的，则group状态为Empty,协议也为None，
        // 否则状态为CompletingRebalance
        //关于分配协议的选择逻辑：找到所有Member都支持的分配协议, 然后让每个Member 根据自己的partition.assignment.strategy配置顺序
        // 找到第一个所有人都支持的的协议投出一票
        // 然后最后比较这些协议, 哪个投票数最多就当选为最终使用的分配协议。
        //中心思想就是：1. 找到所有人都支持的协议  2. 1中的协议在partition.assignment.strategy中更靠前
        group.initNextGeneration()
        if (group.is(Empty)) {
          info(s"Group ${group.groupId} with generation ${group.generationId} is now empty " +
            s"(${Topic.GROUP_METADATA_TOPIC_NAME}-${partitionFor(group.groupId)})")
    // 如果Group是空的话，我们则在 __consumer_offset 里面写入一条Group的元数据,状态是Empty
          groupManager.storeGroup(group, Map.empty, error => {
            if (error != Errors.NONE) {
              warn(s"Failed to write empty metadata for group ${group.groupId}: ${error.message}")
              warn(s"onCompleteJoin 因为当前Group中的member位空了，所以要给Group中写入的空元信息，但是写入失败了, 不过问题也不大,只是警告一下 group ${group.groupId}: ${error.message}")
            }
          })
        } else {
          info(s"Stabilized group ${group.groupId} generation ${group.generationId} " +
            s"(${Topic.GROUP_METADATA_TOPIC_NAME}-${partitionFor(group.groupId)})")
          // trigger the awaiting join group response callback for all the members after rebalancing
          // 当重平衡后触发所有等待Join group的members的的回调接口
          for (member <- group.allMemberMetadata) {
            val joinResult = JoinGroupResult(
              // 如果这个member是leader的话，则传入所有members
              members = if (group.isLeader(member.memberId)) {
                group.currentMemberMetadata
              } else {
                List.empty
              },
              memberId = member.memberId,
              generationId = group.generationId,
              protocolType = group.protocolType,
              protocolName = group.protocolName,
              leaderId = group.leaderOrNull,
              error = Errors.NONE)
            group.maybeInvokeJoinCallback(member, joinResult)
            completeAndScheduleNextHeartbeatExpiration(group, member)
            member.isNew = false
          }
        }
      }
    }
  }

partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor,org.apache.kafka.clients.consumer.RangeAssignor

[2022-09-08 14:34:12,508] INFO [Consumer clientId=client2, groupId=consumer0] Rebalance failed. (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
org.apache.kafka.common.errors.InconsistentGroupProtocolException: The group member's supported protocols are incompatible with those of existing members or first group member tried to join with empty protocol type or empty protocol list.
[2022-09-08 14:34:12,511] ERROR Error processing message, terminating consumer process:  (kafka.tools.ConsoleConsumer$)

private class JoinGroupResponseHandler extends CoordinatorResponseHandler<JoinGroupResponse, ByteBuffer> {
        private JoinGroupResponseHandler(final Generation generation) {
            super(generation);
        }
        @Override
        public void handle(JoinGroupResponse joinResponse, RequestFuture<ByteBuffer> future) {
            Errors error = joinResponse.error();
            if (error == Errors.NONE) {
                if (isProtocolTypeInconsistent(joinResponse.data().protocolType())) {
                    log.error("JoinGroup failed: Inconsistent Protocol Type, received {} but expected {}",
                        joinResponse.data().protocolType(), protocolType());
                    future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
                } else {
                    log.debug("Received successful JoinGroup response: {}", joinResponse);
                    sensors.joinSensor.record(response.requestLatencyMs());
                    synchronized (AbstractCoordinator.this) {
                        if (state != MemberState.PREPARING_REBALANCE) {
                        } else {
                            state = MemberState.COMPLETING_REBALANCE;
                            // 启动心跳线程
                            if (heartbeatThread != null)
                                heartbeatThread.enable();
                            AbstractCoordinator.this.generation = new Generation(
                                joinResponse.data().generationId(),
                                joinResponse.data().memberId(), joinResponse.data().protocolName());
                            log.info("Successfully joined group with generation {}", AbstractCoordinator.this.generation);
                            if (joinResponse.isLeader()) {
                                //如果是Leader Member的话，则需要计算一下新的分配方式,再发起SyncGroup请求
                                //并且会把新的分配数据给带上
                                onJoinLeader(joinResponse).chain(future);
                            } else {
                               //如果是Follower Member的话，直接发起SyncGroup请求
                                onJoinFollower().chain(future);
                            }
                        }
                    }
                }
            } else {
               // 省略....
            }
        }
    }