基于Flink Runner的无状态Beam Pipeline：Pubsublite消息提前ACK致数据丢失

核心误解纠正 #

Checkpoint/Savepoint绝非仅适用于有状态流处理，它的核心作用是记录作业的全局处理进度——包括数据源的消费偏移量、输出端的提交状态等，是实现端到端Exactly-Once语义的基础，无状态的转换+输出场景同样需要依赖它来避免数据丢失。

实现思路与配置方案 #

1. 启用Beam作业的Checkpoint机制 #

通过FlinkPipelineOptions配置Checkpoint，确保Flink Runner会定期生成Checkpoint，只有当Checkpoint成功完成时，才会向Pubsublite提交消息ACK，同时确认Kafka的写入状态。

// 配置PipelineOptions
FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
// 开启Checkpoint，设置间隔（示例为10秒）
options.setCheckpointingInterval(10000);
// 设置Checkpoint模式为EXACTLY_ONCE，保证端到端的精确一次语义
options.setCheckpointMode(CheckpointMode.EXACTLY_ONCE);
// 配置Checkpoint超时时间，避免长时间阻塞
options.setCheckpointTimeout(60000);

2. 配置KafkaIO的Exactly-Once写入 #

Beam的KafkaIO默认是AT_LEAST_ONCE语义，需要开启事务性写入，绑定Checkpoint生命周期，确保只有Checkpoint成功时，Kafka的写入才会被提交。同时要保证两个Kafka主题的写入都参与Checkpoint的一致性校验。

修改你的代码，为每个KafkaIO添加事务配置：

// 处理第一个Kafka主题写入
msgs.apply("Map to ProducerRecord", MapElements.via(new FormatPubSubMessage(options.getPrimaryTopic())))
    .setCoder(ProducerRecordCoder.of(VoidCoder.of(), ByteArrayCoder.of()))
    .apply("Write to Primary Kafka", KafkaIO.<Void, byte[]>writeRecords()
            .withBootstrapServers(options.getBootstrapServers())
            .withTopic(options.getPrimaryTopic())
            .withKeySerializer(VoidSerializer.class)
            .withValueSerializer(ByteArraySerializer.class)
            // 开启事务，绑定Checkpoint，确保写入仅在Checkpoint完成后提交
            .withTransactionalIdPrefix("beam-kafka-primary-" + options.getJobName())
            .withExactlyOnce(true)
    );

// 处理第二个Kafka主题写入（可复用转换逻辑，避免重复解析）
msgs.apply("Map to ProducerRecord for Secondary", MapElements.via(new FormatPubSubMessage(options.getSecondaryTopic())))
    .setCoder(ProducerRecordCoder.of(VoidCoder.of(), ByteArrayCoder.of()))
    .apply("Write to Secondary Kafka", KafkaIO.<Void, byte[]>writeRecords()
            .withBootstrapServers(options.getBootstrapServers())
            .withTopic(options.getSecondaryTopic())
            .withKeySerializer(VoidSerializer.class)
            .withValueSerializer(ByteArraySerializer.class)
            .withTransactionalIdPrefix("beam-kafka-secondary-" + options.getJobName())
            .withExactlyOnce(true)
    );

3. 配置Pubsublite源的确认策略 #

当Beam启用Checkpoint后，Pubsublite源会自动将消息ACK与Checkpoint绑定——只有当Checkpoint成功完成（所有Kafka写入都确认成功），才会向Pubsublite提交消费偏移量，避免提前ACK导致的消息丢失。

如果需要更精细的控制，可以显式配置：

PubsubliteIO.read()
        .fromSubscription(options.getPubsubliteSubscription())
        // 确保源端偏移量提交与Checkpoint绑定
        .withCommitOffsetInCheckpoint(true);

4. 优化Kafka写入的重试机制 #

为Kafka Producer配置本地重试，减少Checkpoint回滚的频率：

Map<String, Object> producerConfigs = new HashMap<>();
// 设置重试次数
producerConfigs.put(ProducerConfig.RETRIES_CONFIG, 10);
// 设置重试间隔
producerConfigs.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000);
// 设置批量发送的确认级别为all，保证写入可靠性
producerConfigs.put(ProducerConfig.ACKS_CONFIG, "all");

// 在KafkaIO中应用配置
KafkaIO.<Void, byte[]>writeRecords()
        // ...其他配置
        .withProducerConfigUpdates(producerConfigs);

关键原理说明 #

• 当Checkpoint触发时，Flink会暂停作业处理，记录所有数据源的当前偏移量，然后等待所有输出端（Kafka写入）完成提交准备。
• 只有当所有输出端都确认可以提交（比如Kafka事务已准备好），Checkpoint才会成功，此时Pubsublite的偏移量会被提交（消息ACK），Kafka的事务会被提交。
• 如果Kafka写入失败，Checkpoint会失败，Flink会回滚到上一个成功的Checkpoint，重新处理这段时间的消息，直到写入成功。

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