在消费者调用poll拉消息的时候，消费者会先检测当前是否需要执行分区再分配操作，如果需要则直接返回空的结果，这样在不超时的情况下，方法 KafkaConsumer#pollOnce 会立即被再次调用，从而开始对当前 topic 分区执行再分配，即调用 ConsumerCoordinator#poll 方法。
public boolean poll(Timer timer) {maybeUpdateSubscriptionMetadata();// 触发执行注册的监听 offset 提交完成的方法invokeCompletedOffsetCommitCallbacks();// 确保当前是 AUTO_TOPICS 或 AUTO_PATTERN（USER_ASSIGNED 不需要再平衡）订阅模式，// 且目标 GroupCoordinator 节点可达，如果不可达，则会尝试寻找一个可用的节点if (subscriptions.partitionsAutoAssigned()) {if (protocol == null) {throw new IllegalStateException("User configured " + ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG +" to empty while trying to subscribe for group protocol to auto assign partitions");}// Always update the heartbeat last poll time so that the heartbeat thread does not leave the// group proactively due to application inactivity even if (say) the coordinator cannot be found.pollHeartbeat(timer.currentTimeMs());if (coordinatorUnknown() && !ensureCoordinatorReady(timer)) {return false;}// 需要执行再平衡if (rejoinNeededOrPending()) {// due to a race condition between the initial metadata fetch and the initial rebalance,// we need to ensure that the metadata is fresh before joining initially. This ensures// that we have matched the pattern against the cluster's topics at least once before joining.if (subscriptions.hasPatternSubscription()) {// For consumer group that uses pattern-based subscription, after a topic is created,// any consumer that discovers the topic after metadata refresh can trigger rebalance// across the entire consumer group. Multiple rebalances can be triggered after one topic// creation if consumers refresh metadata at vastly different times. We can significantly// reduce the number of rebalances caused by single topic creation by asking consumer to// refresh metadata before re-joining the group as long as the refresh backoff time has// passed.if (this.metadata.timeToAllowUpdate(timer.currentTimeMs()) == 0) {this.metadata.requestUpdate();}if (!client.ensureFreshMetadata(timer)) {return false;}maybeUpdateSubscriptionMetadata();}/** 1. 检查目标 GroupCoordinator 节点是否准备好接收请求* 2. 启动心跳线程* 3. 执行分区再分配操作*/// ensureActiveGroup 执行具体的分区再分配操作if (!ensureActiveGroup(timer)) {return false;}}} else {// For manually assigned partitions, if there are no ready nodes, await metadata.// If connections to all nodes fail, wakeups triggered while attempting to send fetch// requests result in polls returning immediately, causing a tight loop of polls. Without// the wakeup, poll() with no channels would block for the timeout, delaying re-connection.// awaitMetadataUpdate() initiates new connections with configured backoff and avoids the busy loop.// When group management is used, metadata wait is already performed for this scenario as// coordinator is unknown, hence this check is not required.if (metadata.updateRequested() && !client.hasReadyNodes(timer.currentTimeMs())) {client.awaitMetadataUpdate(timer);}}// 异步提交 offsetmaybeAutoCommitOffsetsAsync(timer.currentTimeMs());return true;
}当我们使用 AUTO_TOPICS 或 AUTO_PATTERN 模式订阅 Kafka topic 时，我们并不需要考虑当前消费者具体消费哪个分区，Kafka 会依据分区分配策略为消费者分配一个或多个分区进行消费（一个分区至多被一个消费者消费，不允许多个消费者同时消费同一个分区）。但是消费者可能会中途加入，也可能会中途退出，topic 的分区数目也是允许改变的，此时就需要依赖分区再分配机制为注册的消费者重新分配分区。
分区再分配操作分为 3 个阶段，并且是一个与集群交互联动的过程，这里我们以客户端视角，当消费者检测到需要重新分配分区时会触发执行：
- 发送 GroupCoordinatorRequest 请求获取目标可用的 GroupCoordinator 实例所在的 broker 节点，如果没有则选择负载最小的节点并尝试建立连接；
- 向 GroupCoordinator 实例所在节点发送 JoinGroupRequest 请求申请加入目标 group，GroupCoordinator 实例会在既定时间范围内等待消费者的申请加入请求，如果提前检测到已经接收到 group 名下所有消费者的申请，或者等待时间超时，则会返回 JoinGroupResponse 响应，主要目的是告知谁是新的 Group Leader 消费者，以及最终确定的分区分配策略；
- Group Leader 依据指定的分区分配策略为当前 group 名下的消费者分配分区，并向目标 GroupCoordinator 实例所在节点发送 SyncGroupRequest 请求以告知最终的分区分配结果。
判定需要执行分区再分配操作的条件，位于 ConsumerCoordinator#rejoinNeededOrPending
public boolean rejoinNeededOrPending() {// USER_ASSIGNED 订阅模式不需要执行分区再分配if (!subscriptions.partitionsAutoAssigned())return false;// we need to rejoin if we performed the assignment and metadata has changed;// also for those owned-but-no-longer-existed partitions we should drop them as lost// 再平衡过程中分区数量发生变化if (assignmentSnapshot != null && !assignmentSnapshot.matches(metadataSnapshot)) {requestRejoin();return true;}// we need to join if our subscription has changed since the last join// 消费者 topic 订阅信息发生变化if (joinedSubscription != null && !joinedSubscription.equals(subscriptions.subscription())) {requestRejoin();return true;}// 其它标识需要再平衡的操作，例如分区再分配执行失败、重置年代信息等return super.rejoinNeededOrPending();
}如果判定需要执行分区再分配操作，消费者接下去会调用 AbstractCoordinator#ensureActiveGroup 方法确认所属 group 对应的目标 GroupCoordinator 实例所在节点是否准备好接收请求，如果对应节点不可用，则会发送 GroupCoordinatorRequest 请求查找负载较小且可用的节点，并与之建立连接。接着会调用 AbstractCoordinator#joinGroupIfNeeded 方法开始执行分区再分配策略，实现如下：
boolean joinGroupIfNeeded(final Timer timer) {// 如果需要执行分区再分配，且目前正在进行中while (rejoinNeededOrPending()) {// 再次检查目标 GroupCoordinator 节点是否准备好接收请求if (!ensureCoordinatorReady(timer)) {return false;}// call onJoinPrepare if needed. We set a flag to make sure that we do not call it a second// time if the client is woken up before a pending rebalance completes. This must be called// on each iteration of the loop because an event requiring a rebalance (such as a metadata// refresh which changes the matched subscription set) can occur while another rebalance is// still in progress.// 执行前期准备工作if (needsJoinPrepare) {// need to set the flag before calling onJoinPrepare since the user callback may throw// exception, in which case upon retry we should not retry onJoinPrepare either./** 1. 如果开启了 offset 自动提交，则同步提交 offset* 2. 调用注册的 ConsumerRebalanceListener 监听器的 onPartitionsRevoked 方法* 3. 取消当前消费者的 leader 身份（如果是的话），恢复成为一个普通的消费者*/needsJoinPrepare = false;onJoinPrepare(generation.generationId, generation.memberId);}// 创建并发送 JoinGroupRequest 请求，申请加入目标 groupfinal RequestFuture<ByteBuffer> future = initiateJoinGroup();client.poll(future, timer);if (!future.isDone()) {// we ran out of timereturn false;}// 执行分区分配成功if (future.succeeded()) {Generation generationSnapshot;// Generation data maybe concurrently cleared by Heartbeat thread.// Can't use synchronized for {@code onJoinComplete}, because it can be long enough// and shouldn't block heartbeat thread.// See {@link PlaintextConsumerTest#testMaxPollIntervalMsDelayInAssignment}synchronized (AbstractCoordinator.this) {generationSnapshot = this.generation;}if (generationSnapshot != Generation.NO_GENERATION) {// Duplicate the buffer in case `onJoinComplete` does not complete and needs to be retried.ByteBuffer memberAssignment = future.value().duplicate();onJoinComplete(generationSnapshot.generationId, generationSnapshot.memberId, generationSnapshot.protocol, memberAssignment);// Generally speaking we should always resetJoinGroupFuture once the future is done, but here// we can only reset the join group future after the completion callback returns. This ensures// that if the callback is woken up, we will retry it on the next joinGroupIfNeeded.// And because of that we should explicitly trigger resetJoinGroupFuture in other conditions below.resetJoinGroupFuture();needsJoinPrepare = true;} else {log.info("Generation data was cleared by heartbeat thread. Initiating rejoin.");resetStateAndRejoin();resetJoinGroupFuture();return false;}} else {resetJoinGroupFuture();// 执行分区分配失败，依据失败类型考虑是否重试final RuntimeException exception = future.exception();if (exception instanceof UnknownMemberIdException ||exception instanceof RebalanceInProgressException ||exception instanceof IllegalGenerationException ||exception instanceof MemberIdRequiredException)continue;else if (!future.isRetriable())throw exception;timer.sleep(rebalanceConfig.retryBackoffMs);}}return true;
}