在开始执行分区再分配操作之前需要执行一些前期准备工作，这里使用了 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 partitionsrevokedPartitions = 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);}
}