这部分主要来说明消费者对协议的处理。

各个消费者都可设置partition.assignment.strategy(分区分配策略)，服务端是如何处理的呢？

这块的代码要追溯到joinGroup请求结束，通过前面的源码分析我们知道joinGroup主要是判断是否发起rebalance以及等待其他组成员加入组，而在所有成员加入或者RebalanceTimeout
之后会调用onCompleteJoin方法，代码如下。

def onCompleteJoin(group: GroupMetadata): Unit = {
    //1.1 针对动态成员的处理
    group.inLock {
      group.notYetRejoinedMembers.filterNot(_.isStaticMember) foreach { failedMember =>
        removeHeartbeatForLeavingMember(group, failedMember)
        group.remove(failedMember.memberId)
        group.removeStaticMember(failedMember.groupInstanceId)
      }

      if (group.is(Dead)) {
        info(s"Group ${group.groupId} is dead, skipping rebalance stage")
        //leader没有rejoin且没有member能选，则group.maybeElectNewJoinedLeader返回false，我们需要再次延时。通过maybeElectNewJoinedLeader选出leader
      } else if (!group.maybeElectNewJoinedLeader() && group.allMembers.nonEmpty) {
        // If all members are not rejoining, we will postpone the completion
        // of rebalance preparing stage, and send out another delayed operation
        // until session timeout removes all the non-responsive members.
        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 {
        //1.2 状态转为CompletingRebalance，投票选择协议，选择票数最多的那个
        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)})")

          groupManager.storeGroup(group, Map.empty, error => {
            if (error != Errors.NONE) {
              // we failed to write the empty group metadata. If the broker fails before another rebalance,
              // the previous generation written to the log will become active again (and most likely timeout).
              // This should be safe since there are no active members in an empty generation, so we just warn.
              warn(s"Failed to write empty metadata for group ${group.groupId}: ${error.message}")
            }
          })
        } else {
          //1.3 选举完成后就返回
          // trigger the awaiting join group response callback for all the members after rebalancing
          for (member <- group.allMemberMetadata) {
            val joinResult = JoinGroupResult(
              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
          }
        }
      }
    }
  }

针对动态成员的处理

这是一个与前面关联的点，在前面一篇的案例中我们知道如果是静态成员会拥有更长的session时间，而动态成员则是在断连之后第一次Rebalance时就剔除掉下线的member，处理就是在1.1的
代码中。在这里会过滤掉静态成员，将没有加入组的成员移除掉。

group状态转为CompletingRebalance，投票选举协议 kafka.coordinator.group.GroupMetadata#initNextGeneration

在initNextGeneration方法中可以看到会对generationId自增加一，generationId相当于group的纪元，每次发生Rebalance都会自增。接着是设置protocolName，针对选举协议的部分就是在Some(selectProtocol)中。

  def initNextGeneration() = {
    if (members.nonEmpty) {
      generationId += 1
      protocolName = Some(selectProtocol)
      subscribedTopics = computeSubscribedTopics()
      transitionTo(CompletingRebalance)
    } else {
      generationId += 1
      protocolName = None
      subscribedTopics = computeSubscribedTopics()
      transitionTo(Empty)
    }
    receivedConsumerOffsetCommits = false
    receivedTransactionalOffsetCommits = false
  }

对于selectProtocol这段代码还挺好理解的，就是对所有的member设置的协议投票，取票数最多的协议，这个代码逻辑也与很多资料说的相符，但针对协议的处理只是简单这样吗？
大家可以想象一个case：如果有四个消费者，其中三个都设置的StickyAssignor，而剩的这个设置的CooperativeStickyAssignor，正好剩的这个被选为leader会发生什么呢？
(前面分析过，服务端选择消费者leader也很随意，就是取member的第一个)。

  def selectProtocol: String = {
    if (members.isEmpty)
      throw new IllegalStateException("Cannot select protocol for empty group")

    // select the protocol for this group which is supported by all members
    val candidates = candidateProtocols

    // let each member vote for one of the protocols and choose the one with the most votes
    val votes: List[(String, Int)] = allMemberMetadata
      .map(_.vote(candidates))
      .groupBy(identity)
      .mapValues(_.size)
      .toList

    votes.maxBy(_._2)._1
  }

带着上面的疑问我们来测试一下

• 准备三个消费者：
  消费者1：设置StickyAssignor
  消费者2：设置StickyAssignor
  消费者3：设置CooperativeStickyAssignor
• 测试结果:
  消费者3会收到一个异常

  Exception in thread “main” 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.

• 抛出此异常的代码如下，memberProtocolType就是customer，memberProtocols即为消费者设置的protocols，memberProtocols.exists(supportedProtocols
  (_) == members.size)是scala的语法，supportedProtocols是Map结构，key为protocols，value为member中对应protocol
  的个数，这段代码意思是说如果memberProtocols中存在supportedProtocols中包含，且对应的value值等于member的个数，则为true，反之为false
  。也就是说如果第一个加入组的消费者设置了两个协议，则这两个协议的value值都为1，而第二个加入组的消费者就必须要包含其中一个协议，给其中一个协议的value加1，然后第三个消费者也必须包含前面加1
  的那个协议，否则抛异常。所以，实际上为协议投票并没有前面说的那么民主，在加入组时会针对协议做校验，防止最后选择的leader没有这个协议。刚开始我想的是按照这个设定的话，那每次加入组的消费者都必须包含与member
  大小相等的协议的话，那后面针对协议再投票是不是没有必要了？直接取跟memberSize相等的协议不行吗？并不是这样，经过测试同一个消费者还可以为同一个协议投两次票，毕竟protocols的类型是List，然而有一个case
  ，如果第一个消费者只为StickyAssignor投两次票，第二个消费者只投一次，则也会抛InconsistentGroupProtocolException
  ，这种设计看似高大上，实际有种自己跟自己玩然后没啥意思的感觉，我也不继续深究了。

  def supportsProtocols(memberProtocolType: String, memberProtocols: Set[String]) = {
    if (is(Empty))
      !memberProtocolType.isEmpty && memberProtocols.nonEmpty
    else
      protocolType.contains(memberProtocolType) && memberProtocols.exists(supportedProtocols(_) == members.size)
  }

joinGroup返回处理

对于成为leader的消费者，服务端会返回成员信息，其他的则返回空，返回参数样例如下

• 成为leader的消费者

JoinGroupResponseData(throttleTimeMs=0, errorCode=0, generationId=3, protocolType=‘consumer’, protocolName=‘sticky’, leader=‘mykafka-group_4_1-d563db3b-0fcd-4ce8-8e65-12a663dba0f7’, memberId=‘mykafka-group_4_1-d563db3b-0fcd-4ce8-8e65-12a663dba0f7’, members=[JoinGroupResponseMember(memberId=‘mykafka-group_4_2-4c364a51-2d52-446f-b4d4-2b61ae3738c0’, groupInstanceId=null, metadata=[0, 1, 0, 0, 0, 1, 0, 7, 116, 111, 112, 105, 99, 95, 49, -1, -1, -1, -1, 0, 0, 0, 0]), JoinGroupResponseMember(memberId=‘mykafka-group_4_1-d563db3b-0fcd-4ce8-8e65-12a663dba0f7’, groupInstanceId=null, metadata=[0, 1, 0, 0, 0, 1, 0, 7, 116, 111, 112, 105, 99, 95, 49, -1, -1, -1, -1, 0, 0, 0, 0])])

• 其他消费者

JoinGroupResponseData(throttleTimeMs=0, errorCode=0, generationId=3, protocolType=‘consumer’, protocolName=‘sticky’, leader=‘mykafka-group_4_1-d563db3b-0fcd-4ce8-8e65-12a663dba0f7’, memberId=‘mykafka-group_4_2-4c364a51-2d52-446f-b4d4-2b61ae3738c0’, members=[])

收到策略之后消费者是如何处理的呢？

在发送joinGroup请求之后有给response设置处理类JoinGroupResponseHandler，最终返回的是JoinGroupResponseHandler
处理之后的处理结果，我们来看JoinGroupResponseHandler中是如何处理的。

if (error == Errors.NONE) {
    if (isProtocolTypeInconsistent(joinResponse.data().protocolType())) {
        log.debug("JoinGroup failed due to inconsistent Protocol Type, received {} but expected {}",
            joinResponse.data().protocolType(), protocolType());
        future.raise(Errors.INCONSISTENT_GROUP_PROTOCOL);
    } else {
        log.info("Received successful JoinGroup response: {}", joinResponse);
        sensors.joinSensor.record(response.requestLatencyMs());

        synchronized (AbstractCoordinator.this) {
            if (state != MemberState.REBALANCING) {
                // if the consumer was woken up before a rebalance completes, we may have already left
                // the group. In this case, we do not want to continue with the sync group.
                future.raise(new UnjoinedGroupException());
            } else {
                //根据返回参数的回调来做处理
                AbstractCoordinator.this.generation = new Generation(
                    joinResponse.data().generationId(),
                    joinResponse.data().memberId(), joinResponse.data().protocolName());
                //针对是否是leader分开处理
                if (joinResponse.isLeader()) {
                    onJoinLeader(joinResponse).chain(future);
                } else {
                    onJoinFollower().chain(future);
                }
            }
        }
    }
}

针对leader的处理

根据投票决定的分配规则分配分区，分配结束后发送SyncGroupRequest请求，在方法performAssignment中还会更新subscriptionState中的groupSubscription
及subscription，以及org.apache.kafka.clients.consumer.internals.ConsumerCoordinator中的 assignmentSnapshot以及 
metadataSnapshot

 private RequestFuture<ByteBuffer> onJoinLeader(JoinGroupResponse joinResponse) {
        try {
            // perform the leader synchronization and send back the assignment for the group
            //根据分区分配策略来分配组成员处理的分区
            Map<String, ByteBuffer> groupAssignment = performAssignment(joinResponse.data().leader(), joinResponse.data().protocolName(),
                    joinResponse.data().members());

            List<SyncGroupRequestData.SyncGroupRequestAssignment> groupAssignmentList = new ArrayList<>();
            for (Map.Entry<String, ByteBuffer> assignment : groupAssignment.entrySet()) {
                groupAssignmentList.add(new SyncGroupRequestData.SyncGroupRequestAssignment()
                        .setMemberId(assignment.getKey())
                        .setAssignment(Utils.toArray(assignment.getValue()))
                );
            }

            SyncGroupRequest.Builder requestBuilder =
                    new SyncGroupRequest.Builder(
                            new SyncGroupRequestData()
                                    .setGroupId(rebalanceConfig.groupId)
                                    .setMemberId(generation.memberId)
                                    .setProtocolType(protocolType())
                                    .setProtocolName(generation.protocolName)
                                    .setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null))
                                    .setGenerationId(generation.generationId)
                                    .setAssignments(groupAssignmentList)
                    );
            log.debug("Sending leader SyncGroup to coordinator {} at generation {}: {}", this.coordinator, this.generation, requestBuilder);
            return sendSyncGroupRequest(requestBuilder);
        } catch (RuntimeException e) {
            return RequestFuture.failure(e);
        }
    }

针对follower的处理

针对follower就直接发送SyncGroupRequest

private RequestFuture<ByteBuffer> onJoinFollower() {
    // send follower's sync group with an empty assignment
    SyncGroupRequest.Builder requestBuilder =
            new SyncGroupRequest.Builder(
                    new SyncGroupRequestData()
                            .setGroupId(rebalanceConfig.groupId)
                            .setMemberId(generation.memberId)
                            .setProtocolType(protocolType())
                            .setProtocolName(generation.protocolName)
                            .setGroupInstanceId(this.rebalanceConfig.groupInstanceId.orElse(null))
                            .setGenerationId(generation.generationId)
                            .setAssignments(Collections.emptyList())
            );
    log.debug("Sending follower SyncGroup to coordinator {} at generation {}: {}", this.coordinator, this.generation, requestBuilder);
    return sendSyncGroupRequest(requestBuilder);
}

总结

这部分到发送同步数据请求这里就结束了，下一篇会来继续分析发送同步请求之后做的事，以及针对四个分配规则来深入分析