基于Pulsar的Flink流处理如何规避部分失败后的重复处理？

你的核心问题是Flink全局Checkpoint机制导致部分Key失败后，整个作业重启会重复处理已成功的Key。以下是针对你场景的可行解决方案：

一、核心思路：避免作业全局失败，分流处理异常Key #

Flink默认的Checkpoint是全局一致性的，只要有算子失败，整个Checkpoint就会失败，作业重启后会从上一次成功的Checkpoint全量恢复，这必然导致重复处理。所以要避免抛出全局异常，而是将失败的Key消息分流到死信队列（DLQ），让主流程正常完成Checkpoint并提交Pulsar偏移量，成功的Key不会被重复处理，失败的Key后续单独处理。

二、具体实现方案：侧输出流（Side Output）+ Pulsar死信Topic #

1. 定义侧输出标签 #

用于标记失败的消息流：

private static final OutputTag<YourMessageType> FAILED_KEYS_TAG = new OutputTag<YourMessageType>("failed-keys") {};

2. 在窗口/处理算子中捕获异常，分流失败消息 #

在步骤3（存储S3）或步骤4（通知下游）的处理逻辑中捕获异常，将失败的消息发送到侧输出流，不要抛出异常：

public class KeyedProcessHandler extends KeyedProcessFunction<String, YourMessageType, YourMessageType> {
    private static final Logger LOG = LoggerFactory.getLogger(KeyedProcessHandler.class);

    @Override
    public void processElement(YourMessageType msg, Context ctx, Collector<YourMessageType> out) {
        try {
            // 步骤3：处理消息并写入S3
            saveToS3(msg);
            // 步骤4：通知下游应用
            notifyDownstream(msg);
            // 正常消息输出到主流
            out.collect(msg);
        } catch (Exception e) {
            // 将失败消息发送到侧输出流
            ctx.output(FAILED_KEYS_TAG, msg);
            // 记录日志，便于排查问题
            LOG.error("处理Key: {} 失败", msg.getKey(), e);
        }
    }

    // 自定义S3存储逻辑
    private void saveToS3(YourMessageType msg) throws Exception {
        // 你的S3存储实现
    }

    // 自定义下游通知逻辑
    private void notifyDownstream(YourMessageType msg) throws Exception {
        // 你的下游通知实现
    }
}

3. 将侧输出流写入Pulsar死信Topic #

在作业主流程中，获取侧输出流并将其写入专门的死信Topic，后续可以单独启动作业消费这个Topic处理失败的记录：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 主处理流
DataStream<YourMessageType> mainStream = env.fromSource(
        PulsarSource.builder()
                .setServiceUrl("pulsar://your-pulsar-cluster:6650")
                .setTopic("persistent://your-tenant/your-namespace/input-topic")
                .setDeserializationSchema(new YourDeserializationSchema())
                .build(),
        WatermarkStrategy.noWatermarks(),
        "Pulsar-Source"
)
.keyBy(YourMessageType::getKey)
.window(TumblingProcessingTimeWindows.of(Time.seconds(30)))
.process(new KeyedProcessHandler());

// 获取失败消息的侧输出流
DataStream<YourMessageType> failedStream = mainStream.getSideOutput(FAILED_KEYS_TAG);

// 将失败消息写入Pulsar死信Topic
failedStream.sinkTo(
        PulsarSink.builder()
                .setServiceUrl("pulsar://your-pulsar-cluster:6650")
                .setTopic("persistent://your-tenant/your-namespace/failed-topic")
                .setSerializationSchema(new YourSerializationSchema())
                .build()
).name("Pulsar-Failed-Sink");

env.execute("Flink-Pulsar-S3-Processing-Job");

4. 配置Checkpoint确保偏移量正常提交 #

确保Checkpoint配置为EXACTLY_ONCE模式，保证主流程成功后Pulsar偏移量正确提交：

// 30秒一次Checkpoint，和窗口周期对齐
env.enableCheckpointing(30000);
CheckpointConfig config = env.getCheckpointConfig();
// 精确一次语义
config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 最小Checkpoint间隔，避免频繁触发
config.setMinPauseBetweenCheckpoints(500);
// Checkpoint超时时间
config.setCheckpointTimeout(60000);
// 保留外部Checkpoint，作业取消后不删除
config.setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

三、关于你提到的其他思路的说明 #

1. 能否仅不对失败消息确认Pulsar偏移量？ #

不行。Pulsar的偏移量是分区级别的，而非Key级别的。一个Pulsar分区中包含多个Key的消息，Flink的Pulsar Consumer只能基于全局Checkpoint提交整个分区的偏移量，无法单独跳过某个Key对应的消息偏移量。

2. Savepoint是否能解决问题？ #

Savepoint主要用于作业版本升级、集群迁移等场景，它是全量状态的快照，恢复时依然会全量重启作业，无法实现仅重跑失败Key的需求，因此不是你的场景的合适解决方案。

四、额外优化建议 #

• 可以给死信Topic的消息添加失败次数、失败原因等元数据，便于后续重试或人工排查。
• 对于可重试的失败（比如S3临时网络故障），可以在KeyedProcessFunction中使用定时器（ctx.timerService().registerProcessingTimeTimer()）实现本地重试，重试失败后再发送到死信队列。

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