Spring Batch KafkaItemReader 偏移量管理與 Step Scope 策略

本文旨在解決 spring batch 中 KafkaitemReader 在非 jvm 重啟情況下重復消費數據的問題。核心在于理解 kafkaItemReader 的狀態管理機制及其與 Spring Bean 生命周期（特別是單例模式）的沖突。通過引入 Spring Batch 的 @StepScope 注解，確保 KafkaItemReader 在每次任務步驟執行時都創建新的實例，從而正確地從 Kafka 消費者組的最新提交偏移量處開始讀取數據，避免重復處理已消費記錄。

Spring Batch KafkaItemReader 的重復消費問題

在使用 Spring Batch 處理 Kafka 數據時，KafkaItemReader 是一個常用的組件，它能夠從 Kafka 主題中讀取記錄。理想情況下，當一個批處理作業通過調度器多次運行時，KafkaItemReader 應該能夠從上次成功處理的偏移量繼續讀取，而不是每次都從頭開始（偏移量 0）。然而，在某些場景下，尤其是在不重啟 JVM 的情況下，我們可能會觀察到 KafkaItemReader 每次啟動都從偏移量 0 開始讀取，導致重復處理數據。

這一現象通常發生在 Spring Batch 作業通過調度器（如 Spring Scheduler）反復觸發，但整個 Spring 應用上下文并未重啟的環境中。盡管 Kafka 的 _consumer_offsets 主題中正確存儲了消費者組的最新偏移量，KafkaItemReader 似乎未能利用這些信息。

問題根源：Bean 的生命周期與狀態共享

KafkaItemReader 是一個有狀態的組件，它需要維護當前讀取的偏移量信息。Spring Batch 框架通過 saveState(true) 配置來支持 ItemReader 的狀態保存和恢復，這通常依賴于 ExecutionContext。同時，KafkaItemReader 內部會根據配置（特別是 partitionOffsets）來決定如何初始化其消費者。當 partitionOffsets 設置為空的 HashMap 時，它會嘗試從 Kafka 消費者組中獲取已提交的偏移量。

然而，當 KafkaItemReader 被定義為一個普通的 Spring Bean（默認是單例 Singleton）時，問題就出現了。在應用程序的整個生命周期內，這個單例 KafkaItemReader 實例只會被創建一次。當調度器反復調用 jobLauncher.run(job, jobParameters); 來啟動新的作業實例時，如果 KafkaItemReader 是單例的，那么：

1. 首次運行： KafkaItemReader 實例被創建，并從 Kafka 獲取最新的已提交偏移量開始消費。
2. 后續運行（不重啟 JVM）： 由于 KafkaItemReader 實例是單例的，它在第一次運行時已經初始化并可能持有內部狀態（例如，上次讀取的偏移量）。當作業再次啟動時，Spring 容器不會創建一個新的 KafkaItemReader 實例，而是重用現有的單例實例。這個單例實例可能不會重新查詢 Kafka 以獲取最新的已提交偏移量，因為它認為自己已經處于一個已知的狀態，或者其內部的消費者客戶端沒有被正確重置，導致它從一個舊的、甚至初始的偏移量開始讀取。

簡而言之，單例 KafkaItemReader 的生命周期與 Spring 應用上下文的生命周期綁定，而非與每次作業執行的生命周期綁定，這導致其狀態無法在每次作業執行時正確地從 Kafka 重新同步。

解決方案：引入 @StepScope 注解

解決此問題的關鍵在于確保 KafkaItemReader 在每次 Spring Batch 作業的步驟 (Step) 執行時都創建一個全新的實例。Spring Batch 提供了 @StepScope 注解來管理這種特殊的 Bean 生命周期。

@StepScope 注解的作用是：

• 延遲實例化： 被 @StepScope 注解的 Bean 不會在 Spring 應用上下文啟動時立即實例化，而是在其所屬的 Step 首次執行時才被實例化。
• 每次 Step 實例化： 對于每個 Step 的執行，Spring Batch 都會創建一個新的 @StepScope Bean 實例。這意味著，如果一個作業包含多個 Step，或者一個 Step 被多次執行（例如，在失敗后重試），那么每次 Step 執行都會得到一個全新的 Bean 實例。
• 隔離狀態： 每個實例都是獨立的，它們的內部狀態不會相互干擾。

通過將 KafkaItemReader 聲明為 @StepScope，我們可以確保在每次作業啟動并進入讀取步驟時，都會有一個全新的 KafkaItemReader 實例被創建。這個新實例將重新執行其初始化邏輯，包括從 Kafka 消費者組中獲取最新的已提交偏移量，從而避免重復消費。

示例代碼：配置 Step-Scoped KafkaItemReader

以下是如何配置一個 step-scoped 的 KafkaItemReader 的示例：

import org.apache.kafka.clients.consumer.ConsumerConfig; 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.springframework.batch.core.configuration.annotation.StepScope;  import java.util.HashMap; import java.util.List; import java.util.Properties;  @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.key.deserializer}")     private String keyDeserializer;      @Value("${kafka.value.deserializer}")     private String valueDeserializer;      @Value("${kafka.max.partition.fetch.bytes}")     private String maxPartitionFetchBytes;      @Value("${kafka.fetch.max.bytes}")     private String fetchMaxBytes;      @Value("${kafka.auto.offset.reset}")     private String autoOffsetReset; // e.g., "latest" or "earliest"      @Value("${kafka.enable.auto.commit}")     private String enableAutoCommit; // should be false for Spring Batch managed offsets      // 假設分區列表是動態的，或者從配置中獲取     // 實際應用中，你可能需要一個服務來獲取主題的分區信息     private List<Integer> partitionsList = List.of(0, 1, 2); // 示例：假設有3個分區      /**      * 配置一個 Step-Scoped 的 KafkaItemReader。      * 每次 Step 運行時都會創建一個新的實例。      */     @Bean     @StepScope // 關鍵：確保每次 Step 運行時都創建一個新的 KafkaItemReader 實例     public ItemReader<byte[]> kafkaItemReader() {         Properties props = new Properties();         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);         props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, maxPartitionFetchBytes);         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, fetchMaxBytes);         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset); // 通常設置為 "latest"         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit); // Spring Batch 管理偏移量時通常為 "false"          KafkaItemReader<String, byte[]> kafkaItemReader = new KafkaItemReaderBuilder<String, byte[]>()                 .partitions(partitionsList) // 指定要讀取的分區                 .consumerProperties(props)                 .name("kafkaItemReader") // 為 reader 指定一個名稱                 .saveState(true) // 允許 Spring Batch 保存和恢復 reader 的狀態                 .topic(topicName)                 .build();          // 關鍵：設置空的 partitionOffsets，讓 reader 從 Kafka 獲取已提交的偏移量         // 因為是 @StepScope，每次新實例都會重新執行此初始化邏輯         kafkaItemReader.setPartitionOffsets(new HashMap<>());          return kafkaItemReader;     }      // ... 其他 Job 和 Step 的配置 }

配置要點：

• @StepScope 注解： 將 @StepScope 注解添加到 kafkaItemReader() 方法上，這是解決問題的核心。
• saveState(true)： 保持此設置為 true。它允許 Spring Batch 在 ExecutionContext 中保存 KafkaItemReader 的內部狀態。當 KafkaItemReader 是 step-scoped 時，這意味著每次 Step 啟動時，一個新的實例會嘗試從 ExecutionContext 恢復狀態。如果 ExecutionContext 中沒有狀態（例如，首次運行或上一個作業實例已完成），它將回退到從 Kafka 獲取偏移量。
• setPartitionOffsets(new HashMap())： 保持此設置。它指示 KafkaItemReader 不要使用硬編碼的偏移量，而是依賴 Kafka 消費者組的機制來確定起始偏移量。結合 @StepScope，每次新的 ItemReader 實例都會執行此邏輯，確保它從 Kafka 獲取最新的已提交偏移量。
• ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG： 對于 Spring Batch，通常建議將其設置為 false。Spring Batch 會在每個 chunk 成功處理后，通過其內部機制（如 ItemWriter 完成寫入后）負責提交偏移量，以確保數據處理的原子性和一致性。

注意事項與最佳實踐

1. GROUP_ID 的一致性： 確保 Kafka 消費者配置中的 GROUP_ID_CONFIG 對于所有作業運行都是一致的。Kafka 通過消費者組 ID 來跟蹤偏移量。
2. AUTO_OFFSET_RESET_CONFIG： 這個配置決定了當消費者組首次啟動或沒有有效偏移量時，從哪里開始讀取。通常設置為 “latest”（從最新記錄開始）或 “earliest”（從最早記錄開始）。在 Spring Batch 中，當 KafkaItemReader 首次初始化并發現沒有可恢復的狀態時，這個配置會生效。
3. Spring Batch 的事務管理： KafkaItemReader 與 Spring Batch 的事務管理和重試機制緊密集成。確保你的 ItemProcessor 和 ItemWriter 是冪等的，以防在重試或失敗恢復時重復處理數據。
4. 分區的指定： 在 KafkaItemReaderBuilder 中使用 .partitions(partitionsList) 允許你指定要讀取的 Kafka 主題分區。這對于精細控制消費者行為非常有用。
5. Reader 的命名： 為 KafkaItemReader 提供一個唯一的 name (.name(“kafkaItemReader”)) 是一個好習慣，尤其是在日志和調試時。

總結

當 Spring Batch 的 KafkaItemReader 在非 JVM 重啟情況下重復消費數據時，問題通常源于 KafkaItemReader Bean 被定義為單例，導致其狀態在多次作業運行之間未能正確重置。通過將 KafkaItemReader 配置為 @StepScope，可以確保每次批處理步驟執行時都創建一個全新的 KafkaItemReader 實例，從而使其能夠正確地從 Kafka 消費者組的最新提交偏移量處開始讀取數據。這是管理 Spring Batch 中有狀態 ItemReader 的關鍵實踐，尤其是在長期運行或調度型批處理應用中。