/**

     * Join the group and return the assignment for the next generation. This function handles both

     * JoinGroup and SyncGroup, delegating to {@link #performAssignment(String, String, Map)} if

     * elected leader by the coordinator.

     * @return A request future which wraps the assignment returned from the group leader

     */

    private RequestFuture<ByteBuffer> sendJoinGroupRequest() {

        if (coordinatorUnknown())

            return RequestFuture.coordinatorNotAvailable();

        // send a join group request to the coordinator

        log.info("(Re-)joining group {}", groupId);

        JoinGroupRequest.Builder requestBuilder = new JoinGroupRequest.Builder(

                groupId,

                this.sessionTimeoutMs,

                this.generation.memberId,

                protocolType(),

                metadata()).setRebalanceTimeout(this.rebalanceTimeoutMs);

        log.debug("Sending JoinGroup ({}) to coordinator {}", requestBuilder, this.coordinator);

        return client.send(coordinator, requestBuilder)

                .compose(new JoinGroupResponseHandler());

  private Generation generation = Generation.NO_GENERATION;

    protected static class Generation {

        public static final Generation NO_GENERATION = new Generation(

                OffsetCommitRequest.DEFAULT_GENERATION_ID,

                JoinGroupRequest.UNKNOWN_MEMBER_ID,

                null);

        public final int generationId;

        public final String memberId;

        public final String protocol;

        public Generation(int generationId, String memberId, String protocol) {

            this.generationId = generationId;

            this.memberId = memberId;

            this.protocol = protocol;

        }

    }　　

private def doJoinGroup(group: GroupMetadata,

                          memberId: String,

                          clientId: String,

                          clientHost: String,

                          rebalanceTimeoutMs: Int,

                          sessionTimeoutMs: Int,

                          protocolType: String,

                          protocols: List[(String, Array[Byte])],

                          responseCallback: JoinCallback) {

    group synchronized {

      if (!group.is(Empty) && (group.protocolType != Some(protocolType) || !group.supportsProtocols(protocols.map(_._1).toSet))) {

        // if the new member does not support the group protocol, reject it

        responseCallback(joinError(memberId, Errors.INCONSISTENT_GROUP_PROTOCOL.code))

      } else if (memberId != JoinGroupRequest.UNKNOWN_MEMBER_ID && !group.has(memberId)) {

        // if the member trying to register with a un-recognized id, send the response to let

        // it reset its member id and retry

        responseCallback(joinError(memberId, Errors.UNKNOWN_MEMBER_ID.code))

      } else {

        group.currentState match {

          case 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

            // joining without the specified member id,

            responseCallback(joinError(memberId, Errors.UNKNOWN_MEMBER_ID.code))

          case PreparingRebalance =>

            if (memberId == JoinGroupRequest.UNKNOWN_MEMBER_ID) {

              addMemberAndRebalance(rebalanceTimeoutMs, sessionTimeoutMs, clientId, clientHost, protocolType, protocols, group, responseCallback)

            } else {

              val member = group.get(memberId)

              updateMemberAndRebalance(group, member, protocols, responseCallback)

            }

          case AwaitingSync =>

            if (memberId == JoinGroupRequest.UNKNOWN_MEMBER_ID) {

              addMemberAndRebalance(rebalanceTimeoutMs, sessionTimeoutMs, clientId, clientHost, protocolType, protocols, group, responseCallback)

            } else {

              val member = group.get(memberId)

              if (member.matches(protocols)) {

                // member is joining with the same metadata (which could be because it failed to

                // receive the initial JoinGroup response), so just return current group information

                // for the current generation.

                responseCallback(JoinGroupResult(

                  members = if (memberId == group.leaderId) {

                    group.currentMemberMetadata

                  } else {

                    Map.empty

                  },

                  memberId = memberId,

                  generationId = group.generationId,

                  subProtocol = group.protocol,

                  leaderId = group.leaderId,

                  errorCode = Errors.NONE.code))

              } else {

                // member has changed metadata, so force a rebalance

                updateMemberAndRebalance(group, member, protocols, responseCallback)

              }

            }

          case Empty | Stable =>

            if (memberId == JoinGroupRequest.UNKNOWN_MEMBER_ID) {

              // if the member id is unknown, register the member to the group

              addMemberAndRebalance(rebalanceTimeoutMs, sessionTimeoutMs, clientId, clientHost, protocolType, protocols, group, responseCallback)

            } else {

              val member = group.get(memberId)

              if (memberId == group.leaderId || !member.matches(protocols)) {

                // force a rebalance if a member has changed metadata or if the leader sends JoinGroup.

                // The latter allows the leader to trigger rebalances for changes affecting assignment

                // which do not affect the member metadata (such as topic metadata changes for the consumer)

                updateMemberAndRebalance(group, member, protocols, responseCallback)

              } else {

                // for followers with no actual change to their metadata, just return group information

                // for the current generation which will allow them to issue SyncGroup

                responseCallback(JoinGroupResult(

                  members = Map.empty,

                  memberId = memberId,

                  generationId = group.generationId,

                  subProtocol = group.protocol,

                  leaderId = group.leaderId,

                  errorCode = Errors.NONE.code))

              }

            }

        }

        if (group.is(PreparingRebalance))

          joinPurgatory.checkAndComplete(GroupKey(group.groupId))

      }

    }

  }　　

  private def addMemberAndRebalance(rebalanceTimeoutMs: Int,

                                    sessionTimeoutMs: Int,

                                    clientId: String,

                                    clientHost: String,

                                    protocolType: String,

                                    protocols: List[(String, Array[Byte])],

                                    group: GroupMetadata,

                                    callback: JoinCallback) = {

    // use the client-id with a random id suffix as the member-id

    val memberId = clientId + "-" + group.generateMemberIdSuffix

    val member = new MemberMetadata(memberId, group.groupId, clientId, clientHost, rebalanceTimeoutMs,

      sessionTimeoutMs, protocolType, protocols)

    member.awaitingJoinCallback = callback

    group.add(member)

    maybePrepareRebalance(group)

    member

  }

  private def maybePrepareRebalance(group: GroupMetadata) {

    group synchronized {

      if (group.canRebalance)

        prepareRebalance(group)

    }

  }

  private def prepareRebalance(group: GroupMetadata) {

    // if any members are awaiting sync, cancel their request and have them rejoin

    if (group.is(AwaitingSync))

      resetAndPropagateAssignmentError(group, Errors.REBALANCE_IN_PROGRESS)

    group.transitionTo(PreparingRebalance)

    info("Preparing to restabilize group %s with old generation %s".format(group.groupId, group.generationId))

    val rebalanceTimeout = group.rebalanceTimeoutMs

    val delayedRebalance = new DelayedJoin(this, group, rebalanceTimeout)

    val groupKey = GroupKey(group.groupId)

    joinPurgatory.tryCompleteElseWatch(delayedRebalance, Seq(groupKey))

  }

  def tryCompleteJoin(group: GroupMetadata, forceComplete: () => Boolean) = {

    group synchronized {

      if (group.notYetRejoinedMembers.isEmpty)

        forceComplete()

      else false

    }

  }

def notYetRejoinedMembers = members.values.filter(_.awaitingJoinCallback == null).toList

  /**

   * Complete existing DelayedHeartbeats for the given member and schedule the next one

   */

  private def completeAndScheduleNextHeartbeatExpiration(group: GroupMetadata, member: MemberMetadata) {

    // complete current heartbeat expectation

    member.latestHeartbeat = time.milliseconds()

    val memberKey = MemberKey(member.groupId, member.memberId)

    heartbeatPurgatory.checkAndComplete(memberKey)

    // reschedule the next heartbeat expiration deadline

    val newHeartbeatDeadline = member.latestHeartbeat + member.sessionTimeoutMs

    val delayedHeartbeat = new DelayedHeartbeat(this, group, member, newHeartbeatDeadline, member.sessionTimeoutMs)

    heartbeatPurgatory.tryCompleteElseWatch(delayedHeartbeat, Seq(memberKey))

  }

 def onExpireHeartbeat(group: GroupMetadata, member: MemberMetadata, heartbeatDeadline: Long) {

    group synchronized {

      if (!shouldKeepMemberAlive(member, heartbeatDeadline))

        onMemberFailure(group, member)

    }

  }

  private def onMemberFailure(group: GroupMetadata, member: MemberMetadata) {

    trace("Member %s in group %s has failed".format(member.memberId, group.groupId))

    group.remove(member.memberId)

    group.currentState match {

      case Dead | Empty =>

      case Stable | AwaitingSync => maybePrepareRebalance(group)

      case PreparingRebalance => joinPurgatory.checkAndComplete(GroupKey(group.groupId))

    }

  }