Spring Batch KafkaItemReader 重复消费：深入理解与StepScope解决方案

import org.springframework.batch.item.ItemReader;
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 org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.batch.core.configuration.annotation.StepScope; // 导入 StepScope

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

@Configuration
public class KafkaBatchConfiguration {

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

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

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

    @Value("${kafka.fetch.bytes}")
    private String fetchBytes;

    // 假设分区列表已知或动态获取
    // 实际应用中，分区列表可能需要通过Kafka Admin API动态获取
    private List<Integer> partitionsList = Arrays.asList(0, 1, 2); // 示例分区

    @Bean
    @StepScope // 关键：将KafkaItemReader声明为StepScope
    public ItemReader<byte[]> kafkaItemReader() {
        Map<String, Object> consumerProperties = new HashMap<>();
        consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.ByteArrayDeserializer.class);
        consumerProperties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, fetchBytes);
        consumerProperties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, fetchBytes);
        consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 如果没有已提交偏移量，从最新开始
        consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // Spring Batch会管理偏移量提交

        KafkaItemReader<String, byte[]> kafkaItemReader = new KafkaItemReaderBuilder<String, byte[]>()
                .partitions(partitionsList) // 指定要消费的分区
                .consumerProperties(consumerProperties)
                .name("kafkaDataItemReader") // 唯一的Reader名称
                .saveState(true) // 允许Spring Batch保存和恢复Reader的状态（当作业重启时）
                .topic(topicName)
                .build();

        // 明确设置空map，确保Reader会从Kafka获取偏移量，而不是使用预设值
        // 在StepScope下，每次都会创建一个新实例，此设置将有效
        kafkaItemReader.setPartitionOffsets(new HashMap<>()); 

        return kafkaItemReader;
    }

    // 其他Job和Step的配置...
    // 例如：
    // @Bean
    // public Job myKafkaProcessingJob(JobRepository jobRepository, Step processKafkaStep) {
    //     return new JobBuilder("myKafkaProcessingJob", jobRepository)
    //             .start(processKafkaStep)
    //             .build();
    // }
    //
    // @Bean
    // public Step processKafkaStep(JobRepository jobRepository, PlatformTransactionManager transactionManager, 
    //                              ItemReader<byte[]> kafkaItemReader, ItemProcessor<byte[], byte[]> processor, ItemWriter<byte[]> writer) {
    //     return new StepBuilder("processKafkaStep", jobRepository)
    //             .<byte[], byte[]>chunk(10, transactionManager)
    //             .reader(kafkaItemReader)
    //             .processor(processor)
    //             .writer(writer)
    //             .build();
    // }
}