图解KafkaConsumer SyncGroupRequest请求流程

// 公众号：石臻臻的杂货铺
    private RequestFuture<ByteBuffer> onJoinLeader(JoinGroupResponse joinResponse) {
        try {
            // 根据分区分配策略 计算分配情况
            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);
        }
    }

public static final Schema[] SCHEMAS = new Schema[] {
        SCHEMA_0,
        SCHEMA_1,
        SCHEMA_2,
        SCHEMA_3,
        SCHEMA_4,
        SCHEMA_5
    };
    public static final Schema SCHEMA_5 =
        new Schema(
            new Field("group_id", Type.COMPACT_STRING, "The unique group identifier."),
            new Field("generation_id", Type.INT32, "The generation of the group."),
            new Field("member_id", Type.COMPACT_STRING, "The member ID assigned by the group."),
            new Field("group_instance_id", Type.COMPACT_NULLABLE_STRING, "The unique identifier of the consumer instance provided by end user."),
            new Field("protocol_type", Type.COMPACT_NULLABLE_STRING, "The group protocol type."),
            new Field("protocol_name", Type.COMPACT_NULLABLE_STRING, "The group protocol name."),
            new Field("assignments", new CompactArrayOf(SyncGroupRequestAssignment.SCHEMA_4), "Each assignment."),
            TaggedFieldsSection.of(
            )
        );

private List<TopicPartition> partitions;
        private ByteBuffer userData;

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);
    }

def handleSyncGroup(groupId: String,
                      generation: Int,
                      memberId: String,
                      protocolType: Option[String],
                      protocolName: Option[String],
                      groupInstanceId: Option[String],
                      groupAssignment: Map[String, Array[Byte]],
                      responseCallback: SyncCallback): Unit = {
    validateGroupStatus(groupId, ApiKeys.SYNC_GROUP) match {
      case Some(error) if error == Errors.COORDINATOR_LOAD_IN_PROGRESS =>
        // 如果协调器还在加载中, 意味着该组需要从JoinGroup重新开始,返回异常：REBALANCE_IN_PROGRESS 给客户端进行处理
        responseCallback(SyncGroupResult(Errors.REBALANCE_IN_PROGRESS))
      case Some(error) => responseCallback(SyncGroupResult(error))
      case None =>
        groupManager.getGroup(groupId) match {
          case None => responseCallback(SyncGroupResult(Errors.UNKNOWN_MEMBER_ID))
          case Some(group) => doSyncGroup(group, generation, memberId, protocolType, protocolName,
            groupInstanceId, groupAssignment, responseCallback)
        }
    }
  }

// 公众号：石臻臻的杂货铺
     //wx: shiyanzu001
  groupManager.storeGroup(group, assignment, (error: Errors) => {
         group.inLock {
                  // 另一个成员可能在我们等待此回调时加入了该组，
                  // 因此我们必须确保我们仍处于 CompletingRebalance 状态并且在它被调用时处于同一代。
                  // 如果我们已经转换到另一个状态，那么什么都不做
                  if (group.is(CompletingRebalance) && generationId == group.generationId) {
                  if (error != Errors.NONE) {
                      // 写入数据的过程出现了异常了； 
                      //1.更新本地缓存，将所有的member分配方式设置为empty，
                      //2:遍历所有member, 把孔的分配方案返回给他们。
                      resetAndPropagateAssignmentError(group, error)
                      maybePrepareRebalance(group, s"error when storing group assignment during SyncGroup (member: $memberId)")
                    } else {
                      // 传播分配方式；1.先更新本地缓存  2.遍历所有member, 把他们各自的分配方案 返回给他们。
                      setAndPropagateAssignment(group, assignment)
                      //装换状态为Stable
                      group.transitionTo(Stable)
                    }
                  }
                }
              })

/**
    * 公众号：石臻臻的杂货铺
    * VX : shiyanzu001
    **/
    private class SyncGroupResponseHandler extends CoordinatorResponseHandler<SyncGroupResponse, ByteBuffer> {
        private SyncGroupResponseHandler(final Generation generation) {
            super(generation);
        }
        @Override
        public void handle(SyncGroupResponse syncResponse,
                           RequestFuture<ByteBuffer> future) {
            Errors error = syncResponse.error();
            if (error == Errors.NONE) {
               if (isProtocolTypeInconsistent(syncResponse.data.protocolType())) {
                   //省略...
                } else {
                    synchronized (AbstractCoordinator.this) {
                        // 因为在JoinGroup完成的时候状态已经变更为了COMPLETING_REBALANCE, 
                        //想要流转为STABLE只能是这个状态的
                        if (!generation.equals(Generation.NO_GENERATION) && state == MemberState.COMPLETING_REBALANCE) {
                            if (protocolNameInconsistent) {
                            //省略...
                            } else {
                                // 一切正常，将状态变更为稳定状态 
                                state = MemberState.STABLE;
                                rejoinNeeded = false;
                                // record rebalance latency
                                lastRebalanceEndMs = time.milliseconds();
                                lastRebalanceStartMs = -1L;
        // 把获取到的assigment 分配数据往上抛出返回给消费者客户端处理
                                future.complete(ByteBuffer.wrap(syncResponse.data.assignment()));
                            }
                        } else {
                              //省略...
                        }
            } else {
                //异常情况 ，需要重新发起 JoinGroup的请求哦
                requestRejoin();
                if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
                    // 省略很多异常情况代码...
                }
            }
        }
    }

/**
 * 公众号：石臻臻的杂货铺
 * wx: shiyanzu001
 **/
    @Override
    protected void onJoinComplete(int generation,
                                  String memberId,
                                  String assignmentStrategy,
                                  ByteBuffer assignmentBuffer) {
  //省略..
    }