Kafka 2.6新功能：消费者主动触发Rebalance

Posted 2020-12-16 huxi2b

Kafka 2.6引入的新功能：消费者能够主动触发Rebalance。一直以来，Rebalance的触发都是由Coordinator来执行的，但有些场景下消费者端能够主动触发Rebalance会很有必要。举个例子，在ConsumerPartitionAssignor接口中有个subscriptionUserData方法可以实现自定义的用户数据。之后在进行Rebalance时，Leader消费者可以根据这些自定义的用户数据执行特定逻辑的分区消费分配方案的制定。目前，倘若这些用户数据发生了变更，Kafka是不会开启Rebalance的。如果这些自定义数据非常影响分配方案的制定，那么唯一的做法就是调用KafkaConsumer的unsubscribe方法，显式地告知Coordinator成员要退出组。这样做的弊端在于，unsubscribe方法会回收当前已分配的分区。

 

针对这种场景，如果消费者能用“不退出组”的方式主动地发起Rebalance去刷新或提交最新的自定义用户数据，那么就可以解决这个场景中碰到的问题了。基于这些考量，社区于2.6引入了这个功能。主要体现在KafkaConsumer中增加了enforceRebalance方法。如下所示：

@Override
    public void enforceRebalance() {
        acquireAndEnsureOpen();
        try {
            if (coordinator == null) {
                throw new IllegalStateException("Tried to force a rebalance but consumer does not have a group.");
            }
            coordinator.requestRejoin();
        } finally {
            release();
        }
    }　　

可见，enforceRebalance方法就是调用requestRejon方法请求Rebalance的，而不像unsubscribe那样用显式离开组的方式触发Rebalance。以下是unsubscribe的源码，可以对比感受一下：

public void unsubscribe() {
        acquireAndEnsureOpen();
        try {
            fetcher.clearBufferedDataForUnassignedPartitions(Collections.emptySet());
            if (this.coordinator != null) {
                this.coordinator.onLeavePrepare();
                this.coordinator.maybeLeaveGroup("the consumer unsubscribed from all topics");
            }
            this.subscriptions.unsubscribe();
            log.info("Unsubscribed all topics or patterns and assigned partitions");
        } finally {
            release();
        }
    }　　

可以看到，unsubscribe的逻辑要“重”得多，它让消费者主动发起离组操作，从而被回收了所有已分配给它的分区。

 

enforceRebalance方法是一个非阻塞方法，而且它本身不会引发Rebalance。它仅仅是将标识是否需要重新加入组的布尔型变量置为true，因此若要Rebalance被触发，还需要显式调用poll方法。常见的使用方式如下：

consumer.enforceRebalance();
consumer.poll(Duration.ofMillis(500));

　

那么，这个功能应该在什么时候被使用呢？通常情况下，你不需要用到这个功能，因为Coordinator会帮你料理Rebalance的事情。使用这个功能有两个条件：

• 你要使用ConsumerPartitionAssignor接口用于执行分配方案的制定
• 你要指定subscriptionUserData并且使用它帮助制定分配方案

如果不同时满足这两个条件，使用enforceRebalance方法的意义就不大了。

 

最后说一下两种场景下调用该方法的处理逻辑。

第一种情况：Rebalance已经发生时调用enforceRebalance方法。这是很可能出现的情况，即consumer调用enforceRebalance时Rebalance已经发生。Consumer可能正在等待Coordinator发生分配方案，或者等待其他成员加入组。这时，我们面临几种选择：1、让当前Rebalance正常完成，之后不再触发一次Rebalance；2、1、让当前Rebalance正常完成，之后再次触发一次Rebalance。很明显，无论哪种都是合理的，没有什么绝对的对错。目前的结论是Kafka把是否再次触发的权利交给用户来决定。当前Rebalance正常完成后，用户可以自行检查接收到的分配方案是否符合期望或者是否使用上了变更过的userData。然后再决定是否重试刚刚的enforceRebalance方法。

 

第二种情况：该consumer不属于消费者组，可能是因为已经被踢出组或者是尚未加入组。如果此时消费者组未处于Rebalance流程，那么此时消费者还没来得及发送最及时的userData，那么下次Rebalance一定可以包含最新的userData。此时可以尝试重试Rebalance。