Beam on Flink切换Kafka数据源后Pipeline不消费消息问题求助

问题根因 #

第一步处理的文件属于有界数据源，所有数据处理完成后，Beam会自动将下游所有算子的水印推进到MAX_WATERMARK，该值会被写入savepoint。从savepoint恢复作业时，下游算子判定已有水印为最大值，所有新输入的Kafka数据的时间戳和对应水印均小于MAX_WATERMARK，不会触发任何窗口计算或状态处理，因此表现为Pipeline不消费消息。

可行解决方案 #

方案一：修改历史数据处理阶段的水印推进逻辑 #

在第一步处理历史数据时，自定义水印生成规则，所有历史数据处理完成后，不将水印推进到MAX_WATERMARK，而是停止在Kafka数据源的起始读取时间的前一个时间点：

• 首先给历史数据记录打上业务时间戳
• 自定义水印策略，设置水印最大推进上限为Kafka起始读取时间减1毫秒
• 生成savepoint时，算子存储的水印为自定义的上限值，恢复后Kafka数据源的水印可以正常继续向后推进

方案二：通过Flink State Processor API修改Savepoint #

使用Flink的State Processor API读取第一步生成的savepoint，手动修改对应状态转换算子的水印元数据，将MAX_WATERMARK替换为你需要的目标时间戳，生成新的savepoint后再启动切换到Kafka的作业即可。
你使用的Flink 1.9.3版本已经兼容State Processor API，操作时需要注意匹配Beam 2.23.0的状态序列化规则，避免状态兼容错误。

方案三：调整Pipeline结构合并历史和实时处理 #

无需分两次启动作业，在同一个Pipeline内先处理S3历史数据，处理完成后自动切换到Kafka实时数据源：

• 使用Beam的组合源能力，按顺序先读取有界的S3文件，再读取无界的Kafka流
• 整个Pipeline的水印会从历史数据到实时数据连续推进，不会出现跳变到MAX_WATERMARK的问题，也不需要做savepoint切换

注意事项 #

• 历史数据的最大业务时间戳需要小于等于Kafka起始读取时间对应的最小数据时间戳，避免数据乱序导致窗口计算异常
• 自定义水印时需要注意Beam 2.23.0与Flink 1.9.3的水印传递逻辑兼容性，避免出现水印不推进的问题

代码优化示例 #

// 给历史数据记录设置业务时间戳和自定义水印
records = records
  .apply("Set event timestamps", WithTimestamps.of(record -> {
      // 替换为你从记录中解析业务时间的逻辑
      return Instant.parse(parseEventTime(record));
  }))
  .apply("Custom watermark strategy", WithWatermarks
      .<String>forBoundedOutOfOrderness(Duration.ofMinutes(5))
      // 水印最大推进到Kafka起始读取时间前1ms，避免到MAX_WATERMARK
      .withWatermarkIdleStrategy((timestamp, watermark) -> 
          Instant.ofEpochMilli(Math.min(watermark.getMillis(), 
              Instant.parse("2021-09-01T09:29:59.999-00:00").getMillis())))
  );

内容的提问来源于stack exchange，提问作者harish