在开始执行分区再分配操作之前需要执行一些前期准备工作，这里使用了 needsJoinPrepare 字段进行控制，如果当前正在执行分区再分配，则 needsJoinPrepare 字段会被标记为 false，以防止重复执行。准备工作的逻辑实现位于 ConsumerCoordinator#onJoinPrepare 方法中，主要做了 3 件事情：
如果开启了 offset 自动提交，则同步提交 offset 到集群。激活注册的 ConsumerRebalanceListener 监听器的 onPartitionsRevoked 方法。取消当前消费者的 Leader 身份（如果是的话），将其恢复成为一个普通的消费者。 protected void onJoinPrepare(int generation, String memberId) { log.debug("Executing onJoinPrepare with generation {} and memberId {}", generation, memberId); // commit offsets prior to rebalance if auto-commit enabled // 如果开启了 offset 自动提交，则同步提交 offset 到集群。 maybeAutoCommitOffsetsSync(time.timer(rebalanceConfig.rebalanceTimeoutMs)); // the generation / member-id can possibly be reset by the heartbeat thread // upon getting errors or heartbeat timeouts; in this case whatever is previously // owned partitions would be lost, we should trigger the callback and cleanup the assignment; // otherwise we can proceed normally and revoke the partitions depending on the protocol, // and in that case we should only change the assignment AFTER the revoke callback is triggered // so that users can still access the previously owned partitions to commit offsets etc. Exception exception = null; final Set<TopicPartition> revokedPartitions; // 激活注册的 ConsumerRebalanceListener 监听器的 onPartitionsRevoked 方法。 if (generation == Generation.NO_GENERATION.generationId && memberId.equals(Generation.NO_GENERATION.memberId)) { revokedPartitions = new HashSet<>(subscriptions.assignedPartitions()); if (!revokedPartitions.isEmpty()) { log.info("Giving away all assigned partitions as lost since generation has been reset," + "indicating that consumer is no longer part of the group"); exception = invokePartitionsLost(revokedPartitions); subscriptions.assignFromSubscribed(Collections.emptySet()); } } else { switch (protocol) { case EAGER: // revoke all partitions revokedPartitions = new HashSet<>(subscriptions.assignedPartitions()); exception = invokePartitionsRevoked(revokedPartitions); subscriptions.assignFromSubscribed(Collections.emptySet()); break; case COOPERATIVE: // only revoke those partitions that are not in the subscription any more. Set<TopicPartition> ownedPartitions = new HashSet<>(subscriptions.assignedPartitions()); revokedPartitions = ownedPartitions.stream() .filter(tp -> !subscriptions.subscription().contains(tp.topic())) .collect(Collectors.toSet()); if (!revokedPartitions.isEmpty()) { exception = invokePartitionsRevoked(revokedPartitions); ownedPartitions.removeAll(revokedPartitions); subscriptions.assignFromSubscribed(ownedPartitions); } break; } } // 取消当前消费者的 Leader 身份（如果是的话），将其恢复成为一个普通的消费者。 isLeader = false; // 将其 SubscriptionState#groupSubscription 字段中记录的所属 group 名下所有消费者订阅的 topic 集合重置为当前消费者自己订阅的 topic 集合。 subscriptions.resetGroupSubscription(); if (exception != null) { throw new KafkaException("User rebalance callback throws an error", exception); } }
