再均衡监听器

Doubao 中英对照 Rebalancing Listeners

ContainerProperties 有一个名为 consumerRebalanceListener 的属性，它接收 Kafka client 的 ConsumerRebalanceListener 接口的一个实现。如果不提供此属性，容器会配置一个日志监听器，以 INFO 级别记录再均衡事件。框架还添加了一个子接口 ConsumerAwareRebalanceListener。以下清单展示了 ConsumerAwareRebalanceListener 接口定义：

public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListener {

    void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

    void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

    void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

    void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

}

请注意，在分区被回收时有两个回调。第一个会被立即调用。第二个会在任何待提交的偏移量提交之后调用。如果你希望在某个外部仓库中维护偏移量，这会很有用，如以下示例所示：

containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {

    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // acknowledge any pending Acknowledgments (if using manual acks)
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // ...
        store(consumer.position(partition));
        // ...
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // ...
        consumer.seek(partition, offsetTracker.getOffset() + 1);
        // ...
    }
});

important

从版本 2.4 开始，添加了一个新方法 onPartitionsLost()（类似于 ConsumerRebalanceLister 中同名的方法）。ConsumerRebalanceLister 上的默认实现只是调用 onPartitionsRevoked。ConsumerAwareRebalanceListener 上的默认实现什么也不做。当为监听器容器提供自定义监听器（任一类型）时，重要的是你的实现不要从 onPartitionsLost 中调用 onPartitionsRevoked。如果你实现了 ConsumerRebalanceListener，则应当覆盖默认方法。这是因为监听器容器会在调用你实现上的该方法之后，从其 onPartitionsLost 的实现中调用自己的 onPartitionsRevoked。如果你的实现委托给默认行为，那么每次 Consumer 在容器的监听器上调用该方法时，onPartitionsRevoked 都会被调用两次。

Kafka 4.0 Consumer 再均衡协议

Spring for Apache Kafka 4.0 支持 Apache Kafka 4.0 的新 consumer 再均衡协议（KIP-848），它通过服务器驱动的增量分区分配来提升性能。这减少了 consumer 组的再均衡停机时间。

要启用新协议，请配置 group.protocol 属性：

spring.kafka.consumer.properties.group.protocol=consumer

请记住，上面的属性是一个 Spring Boot 属性。如果你没有使用 Spring Boot，可能需要手动设置它，如下所示。

或者，以编程方式设置：

Map<String, Object> props = new HashMap<>();
props.put("group.protocol", "consumer");
ConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory<>(props);

新协议与 ConsumerAwareRebalanceListener 无缝协作。由于增量再均衡，onPartitionsAssigned 可能会以较小的分区集被多次调用，这与传统协议典型的单次回调不同。

新协议使用服务器端分区分配，忽略通过 spring.kafka.consumer.partition-assignment-strategy 设置的客户端自定义分配器。如果检测到自定义分配器，会记录一条警告。要使用自定义分配器，请设置 group.protocol=classic（如果你不为 group.protocol 指定值，则这是默认值）。