轉載 | 滑思眉Philip

問題背景

在Apache SeaTunnel 2.3.9版本的Kafka連接器實現中，存在一個潛在的內存溢出風險。當用户配置流式作業從Kafka讀取數據時，即使設置了讀取速率限制(read_limit.rows_per_second)，系統仍可能出現內存持續增長直至OOM(Out Of Memory)的情況。

問題現象

用户在實際部署中觀察到以下現象：

1. 在8核12G內存的SeaTunnel Engine集羣上運行Kafka到HDFS的流式作業
2. 雖然配置了read_limit.rows_per_second=1的速率限制，但內存使用量在5分鐘內從200MB飆升至5GB
3. 停止作業後內存不釋放，恢復作業後內存繼續增長直至OOM

4. 最終導致worker節點重啓

   根本原因分析

   通過代碼審查發現，問題根源在於KafkaSource類的createReader方法中，elementsQueue被初始化為無界隊列：

elementsQueue = new LinkedBlockingQueue<>();

這種實現方式存在兩個關鍵問題：

1. 隊列無界：LinkedBlockingQueue未指定容量，理論上可以無限增長，當生產者速度遠大於消費者速度時，會導致內存持續增長。
2. 速率限制失效：雖然用户配置了read_limit.rows_per_second=1，但該限制並未真正作用於Kafka數據讀取環節，導致數據持續堆積在內存隊列中。

解決方案

社區通過PR#9041修復了此問題，主要改進包括：

1. 引入有界隊列：將LinkedBlockingQueue替換為固定大小的ArrayBlockingQueue
2. 可配置隊列大小：新增queue.size配置參數，允許用户根據實際情況調整
3. 默認安全值：設置DEFAULT_QUEUE_SIZE=1000作為默認隊列容量

核心實現代碼變更如下：

public class KafkaSource {
    private static final String QUEUE_SIZE_KEY = "queue.size";
    private static final int DEFAULT_QUEUE_SIZE = 1000;
    
    public SourceReader<SeaTunnelRow, KafkaSourceSplit> createReader(
            SourceReader.Context readerContext) {
        int queueSize = kafkaSourceConfig.getInt(QUEUE_SIZE_KEY, DEFAULT_QUEUE_SIZE);
        BlockingQueue<RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>>> elementsQueue =
                 new ArrayBlockingQueue<>(queueSize);
        // ...
    }
}

最佳實踐建議

對於使用SeaTunnel Kafka連接器的用户，建議：

1. 升級版本：使用包含此修復的SeaTunnel版本
2. 合理配置：根據業務需求和數據特徵設置適當的queue.size值
3. 監控內存：即使使用有界隊列，仍需監控系統內存使用情況
4. 理解速率限制：read_limit.rows_per_second參數作用於下游處理環節，而非Kafka消費環節

總結

此問題的修復不僅解決了內存溢出風險，還提高了系統的穩定性和可配置性。通過引入有界隊列和可配置參數，用户可以更好地控制系統資源使用，避免因數據積壓導致的OOM問題。這也體現了開源社區通過用户反饋持續改進產品質量的良性循環。