Kafka 消费者客户端心跳请求

        /**    * 公众号：石臻臻的杂货铺    * wx: shiyanzu001    **/    private boolean enabled = false;         private synchronized void startHeartbeatThreadIfNeeded() {        if (heartbeatThread == null) {            heartbeatThread = new HeartbeatThread();            heartbeatThread.start();        }    }

    synchronized RequestFuture<Void> sendHeartbeatRequest() {        log.debug("Sending Heartbeat request with generation {} and member id {} to coordinator {}",            generation.generationId, generation.memberId, coordinator);        HeartbeatRequest.Builder requestBuilder =                new HeartbeatRequest.Builder(new HeartbeatRequestData()                        .setGroupId(rebalanceConfig.groupId)                        .setMemberId(this.generation.memberId)                        .setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null))                        .setGenerationId(this.generation.generationId));        return client.send(coordinator, requestBuilder)                .compose(new HeartbeatResponseHandler(generation));    }    

    public synchronized RequestFuture<Void> maybeLeaveGroup(String leaveReason) {        RequestFuture<Void> future = null;
        // 从 2.3 开始，只有动态成员才会向 broker 发送 LeaveGroupRequest，group.instance.id 有效的 consumer 被视为从不发送 LeaveGroup 的静态成员，成员过期仅受 session timeout 控制        if (isDynamicMember() && !coordinatorUnknown() &&            state != MemberState.UNJOINED && generation.hasMemberId()) {            // this is a minimal effort attempt to leave the group. we do not            // attempt any resending if the request fails or times out.            log.info("Member {} sending LeaveGroup request to coordinator {} due to {}",                generation.memberId, coordinator, leaveReason);            LeaveGroupRequest.Builder request = new LeaveGroupRequest.Builder(                rebalanceConfig.groupId,                Collections.singletonList(new MemberIdentity().setMemberId(generation.memberId))            );
            future = client.send(coordinator, request).compose(new LeaveGroupResponseHandler(generation));            client.pollNoWakeup();        }  // 重置状态为UNJOINED         resetGenerationOnLeaveGroup();
        return future;    }

  def handleHeartbeat(groupId: String,                      memberId: String,                      groupInstanceId: Option[String],                      generationId: Int,                      responseCallback: Errors => Unit): Unit = {    validateGroupStatus(groupId, ApiKeys.HEARTBEAT).foreach { error =>      if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS)        // the group is still loading, so respond just blindly        responseCallback(Errors.NONE)      else        responseCallback(error)      return    }
    groupManager.getGroup(groupId) match {      case None =>        responseCallback(Errors.UNKNOWN_MEMBER_ID)
      case Some(group) => group.inLock {        if (group.is(Dead)) {          // if the group is marked as dead, it means some other thread has just removed the group          // from the coordinator metadata; this is likely that the group has migrated to some other          // coordinator OR the group is in a transient unstable phase. Let the member retry          // finding the correct coordinator and rejoin.          responseCallback(Errors.COORDINATOR_NOT_AVAILABLE)        } else if (group.isStaticMemberFenced(memberId, groupInstanceId, "heartbeat")) {          responseCallback(Errors.FENCED_INSTANCE_ID)        } else if (!group.has(memberId)) {          responseCallback(Errors.UNKNOWN_MEMBER_ID)        } else if (generationId != group.generationId) {          responseCallback(Errors.ILLEGAL_GENERATION)        } else {          group.currentState match {            case Empty =>              responseCallback(Errors.UNKNOWN_MEMBER_ID)
            case CompletingRebalance =>              // consumers may start sending heartbeat after join-group response, in which case              // we should treat them as normal hb request and reset the timer              val member = group.get(memberId)              completeAndScheduleNextHeartbeatExpiration(group, member)              responseCallback(Errors.NONE)
            case PreparingRebalance =>                val member = group.get(memberId)                completeAndScheduleNextHeartbeatExpiration(group, member)                responseCallback(Errors.REBALANCE_IN_PROGRESS)
            case Stable =>                val member = group.get(memberId)                completeAndScheduleNextHeartbeatExpiration(group, member)                responseCallback(Errors.NONE)
            case Dead =>              throw new IllegalStateException(s"Reached unexpected condition for Dead group $groupId")          }        }      }    }  }  

  def onExpireHeartbeat(group: GroupMetadata, memberId: String, isPending: Boolean): Unit = {    group.inLock {      if (group.is(Dead)) {    // 如果当前心跳监测到Group协调器已经Dead了，仅仅只是打印一下日志, 因为它自身可能已经不是组协调器了，他已经不能再被允许做什么了        info(s"Received notification of heartbeat expiration for member $memberId after group ${group.groupId} had already been unloaded or deleted.")        } else if (isPending) {        info(s"Pending member $memberId in group ${group.groupId} has been removed after session timeout expiration.")  // 客户端在发起第一次JoinGroup请求的时候，并没有带上memberId,但是Group会生成一个给客户端返回  // 这个时候这个member就是Pending状态的,属于待加入状态, 因为它还会发起第二次JoinGroup请求并带上这个memberId，才算是真的Join了Group  // 在这里 直接把这个memberId从Pending缓存中移除了,因为它心跳监测过期了,这意味着客户端需要重新发起第一次Join        removePendingMemberAndUpdateGroup(group, memberId)      } else if (!group.has(memberId)) {        debug(s"Member $memberId has already been removed from the group.")      } else {        val member = group.get(memberId)        if (!member.hasSatisfiedHeartbeat) {          info(s"Member ${member.memberId} in group ${group.groupId} has failed, removing it from the group")
          removeMemberAndUpdateGroup(group, member, s"removing member ${member.memberId} on heartbeat expiration")        }      }    }  }