Flink Kafka作业设置Boundedness为latest offset时运行异常问题咨询

该问题核心是Flink Kafka EXACTLY_ONCE语义的事务实现机制、Kafka消费者隔离级别、有界KafkaSource的终止判断逻辑三者共同作用的结果，结合Ververica Platform与本地运行环境的配置差异触发：

• Flink Kafka生产者开启EXACTLY_ONCE语义时，所有写入消息都会封装在Kafka事务中，仅当Flink完成checkpoint、事务正式提交后，消息才会对配置了read_committed隔离级别的消费者可见。此时消费者可见的最大偏移量为最后稳定偏移量（LSO），远小于未提交事务对应的最高水位（HW）。
• 你配置的setBounded(OffsetsInitializer.latest())逻辑为：KafkaSource启动时会拉取一次所有分区的latest偏移量作为终止边界，消费到该偏移量后作业自动结束。
• 环境配置差异是触发问题的关键：
  • 本地IDE运行时，消费者默认隔离级别为read_uncommitted，可以读到未提交的事务消息，启动时拉取的latest偏移量为真实的HW，消费到对应偏移量后作业正常结束。
  • Ververica Platform默认会给Kafka消费者强制配置isolation.level = read_committed，同时平台上运行的Flink作业checkpoint间隔通常比本地测试长很多，若消费者启动时生产者的写入事务还未完成提交，拉取到的latest偏移量为远小于HW的LSO，会出现两种异常：如果LSO等于消费起始偏移量，作业会以为没有数据可消费直接挂起；如果后续事务提交，source也不会更新终止偏移量，导致新提交的消息无法被消费。
• 额外验证逻辑：如果生产者使用AT_LEAST_ONCE语义，不会开启Kafka事务，所有消息写入后直接可见，不存在可见性延迟问题，因此作业运行正常。

• 方案1：调整作业启动时序
  如果第一个写入Kafka的作业是单次运行的批作业，等它完全运行结束、所有事务都提交完成后，再启动第二个计数作业，此时消费者拉取的latest偏移量就是所有已提交消息的最大偏移量，作业运行符合预期。
• 方案2：调整生产者事务配置
  检查Ververica Platform上生产者作业的配置：
  1. 确保checkpoint功能正常开启，没有连续checkpoint失败的情况，保证事务可以正常提交
  2. 确保Kafka生产者的transaction.timeout.ms配置大于作业的checkpoint间隔+checkpoint保留时间，避免事务超时回滚
• 方案3：调整消费者配置（仅适用允许非强一致的场景）
  显式将消费者的isolation.level设置为read_uncommitted，可以读取到未提交的事务消息，和本地运行逻辑一致，但会损失一致性保证，可能读到回滚的脏数据。
• 方案4：调整有界源终止逻辑
  如果需要在生产者运行同时启动消费作业，可以替换setBounded(OffsetsInitializer.latest())的逻辑，改为自定义终止条件，比如固定消费时长、监听生产者的结束标识消息等，避免依赖启动时单次拉取的偏移量。

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