Flink 1.15.1 Kafka Sink异常事务ID错误排查问询

问题场景 #

Flink 1.15.1作业配置了execution.checkpointing.mode='EXACTLY_ONCE'，使用KafkaSinkBuilder构建Sink时未调用setDeliverGuarantee（预期默认使用NONE交付保障），也未设置transactionalIdPrefix。第一次触发检查点后作业失败，报错如下：

Sink: Committer (2/2)#732 (36640a337c6ccdc733d176b18adab979) switched from INITIALIZING to FAILED with failure cause: java.lang.IllegalStateException: Failed to commit KafkaCommittable{producerId=4521984, epoch=0, transactionalId=}
...
Caused by: org.apache.kafka.common.config.ConfigException: Invalid value for configuration transactional.id: String must be non-empty

临时将检查点模式改为AT_LEAST_ONCE后问题规避，但需明确错误根本原因。

根本原因分析 #

Flink 1.15.x的KafkaSink存在全局检查点模式驱动的交付保障自动升级逻辑：
当作业全局检查点模式设置为EXACTLY_ONCE时，若KafkaSink未显式指定交付保障，Flink会自动将其交付保障提升为EXACTLY_ONCE——这是为了满足端到端EXACTLY_ONCE语义的一致性要求，强制所有Sink参与事务或两阶段提交流程。

而KafkaSink启用EXACTLY_ONCE交付保障时，必须通过transactionalIdPrefix生成合法的事务ID。由于未配置该前缀，Kafka客户端尝试使用空字符串作为transactional.id，触发了Kafka的配置校验规则（要求transactional.id非空），最终导致报错。

解决方案 #

根据业务需求选择以下方案：

• 需要端到端EXACTLY_ONCE语义：在KafkaSinkBuilder中显式配置：

  KafkaSink.<String>builder()
      .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
      .setTransactionalIdPrefix("your-job-id-sink-prefix") // 自定义非空前缀
      // 其他配置
      .build();
  

• 不需要KafkaSink参与事务：
  1. 保持作业检查点模式为AT_LEAST_ONCE；
  2. 或显式将KafkaSink的交付保障设置为NONE/AT_LEAST_ONCE，即使全局检查点为EXACTLY_ONCE，Sink也不会启用事务（注意：此方式无法保证端到端EXACTLY_ONCE，可能出现重复写入）：

     KafkaSink.<String>builder()
         .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
         // 其他配置
         .build();
     

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