如何在不重启Spring Kafka容器的情况下手动触发消费者组rebalance

存在可实现「不重启容器、早于metadata max age触发rebalance」的方案，以下是两种常用的落地方式：

方案1：反射调用原生Kafka消费者API #

Kafka 2.6+ 提供的enforceRebalance()是原生消费者的公开方法，只是Spring Kafka没有对外暴露容器持有的消费者实例，可通过反射获取原生实例后直接调用：

import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListenerContainer;
import java.lang.reflect.Method;

// 注入容器注册表，所有@KafkaListener生成的容器都会注册到该实例中
@Autowired
private KafkaListenerEndpointRegistry registry;

public void triggerRebalance(String listenerId) {
    MessageListenerContainer container = registry.getListenerContainer(listenerId);
    if (!(container instanceof ConcurrentMessageListenerContainer)) {
        return;
    }
    ConcurrentMessageListenerContainer<?, ?> concurContainer = (ConcurrentMessageListenerContainer<?, ?>) container;
    // 遍历并发消费场景下的所有子容器
    concurContainer.getContainers().forEach(subContainer -> {
        try {
            // 反射获取子容器持有的原生消费者实例
            Method getConsumerMethod = KafkaMessageListenerContainer.class.getDeclaredMethod("doGetConsumer");
            getConsumerMethod.setAccessible(true);
            Consumer<?, ?> nativeConsumer = (Consumer<?, ?>) getConsumerMethod.invoke(subContainer);
            // 触发rebalance
            nativeConsumer.enforceRebalance();
        } catch (Exception e) {
            // 业务侧自行处理异常
            e.printStackTrace();
        }
    });
}

注意：该方案仅支持Kafka客户端版本>=2.6的场景

方案2：无侵入的订阅重置触发 #

不需要反射、也不限制Kafka版本，利用Kafka消费者「订阅关系变更会自动触发rebalance」的特性实现：

1. 反射获取原生消费者实例后，先执行unsubscribe()取消订阅
2. 再用原来的Topic列表执行subscribe()重新订阅
3. 两步操作完成后会立即触发消费者组的rebalance，不需要等待metadata max age周期

注意事项 #

• 触发rebalance会导致该消费者组所有消费者短时间暂停消费，建议避开业务高峰操作，不要频繁触发
• 手动提交offset的场景，建议触发前先提交一次已消费的offset，降低消息重复/丢失的风险

内容的提问来源于stack exchange，提问作者Ruyik