Spring Batch KafkaItemReader 偏移量管理：避免重复消费的关键策略

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.kafka.KafkaItemReader;
import org.springframework.batch.item.kafka.builder.KafkaItemReaderBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.Arrays;

@Configuration
public class KafkaBatchConfig {

    @Value("${kafka.bootstrap.servers}")
    private String KAFKA_CONFIG_BOOTSTRAP_SERVERS;

    @Value("${kafka.group.id}")
    private String KAFKA_CONFIG_GROUP_ID;

    @Value("${kafka.topic.name}")
    private String KAFKA_TOPIC_NAME;

    // 假设分区列表是动态的，或者从配置中获取
    @Value("${kafka.partitions}")
    private String KAFKA_PARTITIONS; // 例如 "0,1,2"

    // 推荐在Spring Batch中使用手动提交，因此ENABLE_AUTO_COMMIT_CONFIG通常设为false
    // Spring Batch的ItemWriter通常会负责在事务边界提交偏移量
    @Value("${kafka.enable.auto.commit:false}")
    private String KAFKA_CONFIG_ENABLE_AUTO_COMMMIT;

    @Value("${kafka.auto.offset.reset:latest}")
    private String KAFKA_CONFIG_AUTO_OFFSET_RESET;

    @Value("${kafka.max.partition.fetch.bytes:1048576}") // 1MB
    private String KAFKA_CONFIG_MAX_PARTITION_FETCH_BYTES;

    @Value("${kafka.fetch.max.bytes:52428800}") // 50MB
    private String KAFKA_CONFIG_FETCH_MAX_BYTES;

    @Bean
    @StepScope // 关键：将KafkaItemReader定义为StepScope
    public KafkaItemReader<String, byte[]> kafkaItemReader() {
        // 配置Kafka消费者属性
        Map<String, Object> consumerProperties = new HashMap<>();
        consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONFIG_BOOTSTRAP_SERVERS);
        consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_CONFIG_GROUP_ID);
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.ByteArrayDeserializer.class);
        consumerProperties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, KAFKA_CONFIG_MAX_PARTITION_FETCH_BYTES);
        consumerProperties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, KAFKA_CONFIG_FETCH_MAX_BYTES);
        consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, KAFKA_CONFIG_AUTO_OFFSET_RESET);
        consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, KAFKA_CONFIG_ENABLE_AUTO_COMMMIT);

        // 解析分区列表
        List<Integer> partitionsList = Arrays.stream(KAFKA_PARTITIONS.split(","))
                                            .map(Integer::parseInt)
                                            .collect(Collectors.toList());

        KafkaItemReader<String, byte[]> reader = new KafkaItemReaderBuilder<String, byte[]>()
                .partitions(partitionsList) // 指定要消费的分区
                .consumerProperties(consumerProperties)
                .name("kafkaItemReader") // 为ItemReader指定一个名称，用于保存状态
                .saveState(true) // 允许Spring Batch保存和恢复ItemReader的状态
                .topic(KAFKA_TOPIC_NAME)
                .build();

        // 明确设置空Map，指示KafkaItemReader从Kafka中读取偏移量
        // 这在StepScope下尤其重要，确保每次新实例都从Kafka获取
        reader.setPartitionOffsets(new HashMap<>());

        return reader;
    }

    // 假设你有一个Job和Step的配置
    // @Bean
    // public Job myKafkaJob(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    //     return new JobBuilder("myKafkaJob", jobRepository)
    //             .start(myKafkaStep(jobRepository, transactionManager))
    //             .build();
    // }

    // @Bean
    // public Step myKafkaStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
    //     return new StepBuilder("myKafkaStep", jobRepository)
    //             .<String, byte[]>chunk(10, transactionManager) // 每次处理10条记录
    //             .reader(kafkaItemReader())
    //             .processor(itemProcessor()) // 你的ItemProcessor
    //             .writer(itemWriter()) // 你的ItemWriter
    //             .build();
    // }

    // ... 其他ItemProcessor和ItemWriter的Bean定义
}