解决Spring Batch KafkaItemReader重复消费：理解与应用Step Scope

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

import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.Arrays; // For example partitionsList

@Configuration
public class KafkaBatchConfig {

    // Kafka配置属性
    private static final String KAFKA_CONFIG_BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String KAFKA_CONFIG_GROUP_ID = "my-spring-batch-consumer-group";
    private static final String KAFKA_CONFIG_KEY_DESERIALIZER_CLASS = StringDeserializer.class.getName();
    private static final String KAFKA_CONFIG_VALUE_DESERIALIZER_BYTES_CLASS = ByteArrayDeserializer.class.getName();
    private static final String KAFKA_CONFIG_AUTO_OFFSET_RESET_LATEST = "latest";
    private static final String KAFKA_CONFIG_ENABLE_AUTO_COMMMIT = "false"; // Spring Batch通常手动管理偏移量

    // 假设的主题和分区列表
    private static final String TOPIC_NAME = "my-topic";
    private static final List<Integer> PARTITIONS_LIST = Arrays.asList(0, 1, 2); // 示例分区

    /**
     * 定义KafkaItemReader Bean，并应用@StepScope
     * 确保每次步骤执行时都创建一个新的实例，从而正确读取Kafka偏移量。
     */
    @Bean
    @StepScope // 核心：确保每次Step执行时都创建新的KafkaItemReader实例
    public ItemReader<byte[]> kafkaBytesItemReader() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONFIG_BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_CONFIG_GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KAFKA_CONFIG_KEY_DESERIALIZER_CLASS);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KAFKA_CONFIG_VALUE_DESERIALIZER_BYTES_CLASS);
        // 通常为了性能，会设置FETCH_MAX_BYTES_CONFIG或MAX_PARTITION_FETCH_BYTES_CONFIG
        // props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "1048576"); // 1MB
        // props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "52428800"); // 50MB

        // auto.offset.reset设置为latest，表示如果找不到已提交的偏移量，则从最新消息开始消费
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, KAFKA_CONFIG_AUTO_OFFSET_RESET_LATEST);
        // 禁用自动提交，由Spring Batch框架管理偏移量提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, KAFKA_CONFIG_ENABLE_AUTO_COMMMIT);

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

        // 当partitions()方法被调用时，KafkaItemReader会尝试从Kafka中获取已提交的偏移量。
        // 如果没有提供明确的partitionOffsets，它会依赖Kafka的消费者组机制。
        // kafkaItemReader.setPartitionOffsets(new HashMap<>()); // 在StepScope下通常不需要显式设置空Map，因为它会重新初始化并查询Kafka
                                                              // 如果不指定，KafkaItemReader会默认从Kafka的消费者组中读取偏移量。
        return kafkaItemReader;
    }

    // ... 其他Job和Step的配置
}