Spring Batch中KafkaItemReader为何不支持自动重平衡？如何实现？

1. 原生不支持自动重平衡的原因 #

Spring Batch的核心设计目标是批处理的确定性与可重启性。Kafka的自动重平衡会动态变更消费者分配的分区，而Batch作业需要明确的读取边界（分区范围、偏移量）来保证重启时能精准恢复断点。如果依赖自动重平衡，作业的读取范围会变得不可控，可能出现重复读取或数据遗漏，直接违背了批处理的一致性要求。

另外，KafkaItemReader本质是静态规划的批量拉取模式，它需要提前确定分区集合来规划批处理分片和进度追踪，自动重平衡的动态特性和这种静态模式天然不匹配。

2. 解决局限的方案：自定义支持重平衡的ItemReader #

可以基于Kafka原生消费者API，结合Spring Batch的ItemReader接口实现自定义Reader，利用Kafka的消费者重平衡监听器处理分区变更。核心思路：

• 使用Kafka消费者的subscribe()方法订阅主题，开启自动分区分配（而非显式指定分区）
• 通过ConsumerRebalanceListener监听分区分配/撤销事件，动态更新当前读取的分区集合
• 实现偏移量持久化逻辑，结合Spring Batch的ExecutionContext保证作业重启时能从正确位置恢复

3. 代码示例 #

以下是简化的自定义重平衡Kafka ItemReader实现：

import org.springframework.batch.item.ItemReader;
import org.springframework.lang.Nullable;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

public class RebalancingKafkaItemReader<T> implements ItemReader<T>, ConsumerRebalanceListener {

    private final Consumer<String, T> kafkaConsumer;
    private final String topic;
    private Set<TopicPartition> assignedPartitions = new HashSet<>();
    private Iterator<ConsumerRecord<String, T>> recordIterator = Collections.emptyIterator();

    public RebalancingKafkaItemReader(Consumer<String, T> kafkaConsumer, String topic) {
        this.kafkaConsumer = kafkaConsumer;
        this.topic = topic;
        // 订阅主题并绑定重平衡监听器
        this.kafkaConsumer.subscribe(Collections.singletonList(topic), this);
    }

    @Nullable
    @Override
    public T read() throws Exception {
        if (!recordIterator.hasNext()) {
            ConsumerRecords<String, T> records = kafkaConsumer.poll(Duration.ofMillis(100));
            if (records.isEmpty()) {
                return null;
            }
            recordIterator = records.iterator();
        }
        ConsumerRecord<String, T> record = recordIterator.next();
        // 可选：将偏移量保存到ExecutionContext，用于作业重启恢复
        // saveOffset(record.topic(), record.partition(), record.offset());
        return record.value();
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 分区撤销时提交当前偏移量，避免重复消费
        kafkaConsumer.commitSync();
        assignedPartitions.removeAll(partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        assignedPartitions.addAll(partitions);
        // 从之前保存的偏移量位置开始读取（需实现偏移量恢复逻辑）
        // partitions.forEach(tp -> kafkaConsumer.seek(tp, getSavedOffset(tp)));
    }

    // 辅助方法：从ExecutionContext读取保存的偏移量
    // private long getSavedOffset(TopicPartition tp) {
    //     // 实现从ExecutionContext获取偏移量的逻辑
    //     return 0;
    // }

    // 辅助方法：保存偏移量到ExecutionContext
    // private void saveOffset(String topic, int partition, long offset) {
    //     // 实现将偏移量保存到ExecutionContext的逻辑
    // }
}

使用注意事项 #

• 必须实现偏移量的持久化与恢复逻辑，否则作业重启会从分区起始位置重新读取
• 重平衡发生时，需保证当前批次的处理一致性，比如在撤销分区前完成当前批次的提交
• 可将Kafka消费者配置为Spring Bean，通过依赖注入管理其生命周期

内容的提问来源于stack exchange，提问作者Đại Tấn