Flink SQL Upsert-Kafka Sink 异常Tombstone问题排查求助

问题重现 #

基于Flink Table API的SQL作业，从8分区Kafka主题读取Offer数据，通过ROW_NUMBER窗口筛选每个商品的最低价有效报价（quantity > 0），最终写入Upsert-Kafka主题。初始运行正常，但一段时间后，部分明明存在有效最低价的商品会意外生成Tombstone消息，重启作业可临时修复，但问题会复发。

环境配置：

• Flink 1.13.2，Upsert-Kafka作为Source/Sink
• Kafka 2.8，作业并行度与Kafka分区数一致（8）
• RocksDB作为状态后端
• 同一商品的Offer分布在Kafka不同分区

可能原因 #

1. ROW_NUMBER窗口的状态一致性问题
   使用ROW_NUMBER() OVER(PARTITION BY id_item ORDER BY price ASC)时，Flink会为每个id_item维护窗口状态。当某个Offer的quantity被更新为0（触发过滤），原有的n_row=1记录会从状态中移除，但如果新的最低价Offer的事件延迟到达，此时该id_item的状态中无符合条件的记录，Upsert-Kafka会发送Tombstone。若后续新的有效Offer到达时，状态未正确触发重新计算，就会导致Tombstone残留。

2. Flink 1.13.2的已知Bug
   Flink 1.13.2存在多个与Upsert-Kafka连接器、窗口状态管理相关的未修复问题，比如状态清理不彻底、Upsert消息生成逻辑异常等，这些Bug会在作业运行一段时间后触发异常Tombstone。

3. 状态膨胀与清理不及时
   RocksDB状态后端若未配置合理的状态TTL（Time-To-Live），会导致无效的Offer状态持续累积，进而影响窗口计算的正确性，触发异常的主键状态判断。

4. 过滤逻辑与Upsert语义的冲突
   WHERE quantity > 0的过滤会动态排除无效Offer，当某个商品的所有Offer被临时过滤（比如更新事件先到，有效Offer后到），会触发Tombstone；若后续有效Offer的计算未正确覆盖之前的Tombstone状态，就会出现数据异常。

解决方向 #

1. 升级Flink版本
   优先升级到Flink 1.13.5及以上的1.13.x补丁版本，或直接升级到1.14+稳定版本。这些版本修复了大量1.13.2中存在的状态管理和Upsert-Kafka相关Bug，能从根源减少此类问题。

2. 替换ROW_NUMBER为聚合函数实现
   改用MIN(price)结合FIRST_VALUE的聚合逻辑替代ROW_NUMBER窗口，聚合状态的管理比窗口更稳定，避免窗口状态的异常清理问题：

   INSERT INTO cheapest_item_offer
   WITH aggregated_offers AS (
       SELECT
           id_item,
           MIN(price) AS min_price,
           FIRST_VALUE(id_offer) OVER (PARTITION BY id_item ORDER BY price ASC) AS id_offer,
           -- 其他字段根据业务需求使用合适的聚合函数，比如FIRST_VALUE/LAST_VALUE
           FIRST_VALUE(xxx) OVER (PARTITION BY id_item ORDER BY price ASC) AS xxx
       FROM offer
       WHERE quantity > 0
       GROUP BY id_item
   )
   SELECT
       id_offer,
       id_item,
       min_price AS price,
       -- ... 其他字段
   FROM aggregated_offers
   -- ... 后续JOIN逻辑
   

3. 配置状态TTL
   在TableConfig中设置状态TTL，确保无效状态被及时清理，避免状态膨胀：

   TableConfig tableConfig = tableEnv.getConfig();
   tableConfig.setIdleStateRetention(Duration.ofHours(2));
   

   或在SQL中通过窗口水印（如果使用事件时间）辅助状态清理：

   SELECT *,
          ROW_NUMBER() OVER(
            PARTITION BY id_item
            ORDER BY price ASC, event_time DESC
          ) AS n_row
   FROM (
       SELECT *,
              WATERMARK FOR event_time AS event_time - INTERVAL '10' MINUTE
       FROM offer
       WHERE quantity > 0
   )
   

4. 启用调试日志排查
   开启org.apache.flink.connector.kafka.sink的DEBUG级别日志，追踪Tombstone生成的具体时机和对应的id_item，结合事件流日志定位触发异常的具体数据场景。

5. 优化Upsert-Kafka Sink配置
   调整Sink的缓冲配置，避免因批量刷新导致的消息顺序问题：

   'sink.buffer-flush.interval' = '500',
   'sink.buffer-flush.max-rows' = '50'
   

   同时确保主键id_item的哈希分布均匀，避免单并行子任务状态过载。

内容的提问来源于stack exchange，提问作者Jaume Jiménez