发送聚合数据至异步步骤后清理状态是否安全？

我想要构建一款对数据进行聚合处理的应用，通过定时器将聚合结果发送至异步步骤进行转储。在onTimer函数中发送数据后，我会清理状态，代码示例如下：

@Override
public void onTimer(long timestamp, KeyedProcessFunction<KEY, IN, Aggregation>.OnTimerContext ctx, Collector<Aggregation> out) throws Exception {
    out.collect(new Aggregation(ctx.getCurrentKey(), aggregation.get()));
    aggregation.clear();
}

该流会传入AsyncDataStream，代码如下：

SingleOutputStreamOperator<Aggregation> aggregations;
AsyncDataStream.unorderedWaitWithRetry(aggregations, new AsyncDatabaseRequest(), 10, TimeUnit.SECONDS, 1000, asyncRetryStrategy).addSink(new DiscardingSink<>());

请问发送聚合数据至异步步骤后清理状态是否安全？若数据写入目标失败会发生什么？

1. 发送数据后立刻清理状态并不安全 #

Flink的状态一致性依赖检查点机制，这种操作存在明显风险：

• 检查点可能在aggregation.clear()执行后完成快照，此时状态已被清空；
• 若后续异步写入失败触发故障恢复，恢复后的状态是清空后的状态，之前的聚合数据会永久丢失，无法重新生成并发送。
  这种做法直接破坏了Flink的至少一次语义保障。

2. 数据写入目标失败的后果 #

• 重试阶段：unorderedWaitWithRetry会按照配置的asyncRetryStrategy发起重试，比如在指定次数内重复发送异步请求；
• 重试耗尽后：所有重试失败时，Flink会将该错误判定为任务故障，触发算子重启（具体行为取决于集群故障恢复策略）；
• 不可逆数据丢失：由于提前清理了聚合状态，任务重启后无法重新生成之前的聚合结果，这部分数据彻底丢失；
• 任务异常循环：如果重试策略配置不合理（比如无限重试），会导致任务陷入持续重试的死循环，无法处理后续正常数据。

优化建议 #

• 延迟状态清理：不在onTimer中立刻清空状态，而是在异步写入确认成功后，通过回调触发状态清理（需在KeyedContext环境下操作状态）；
• 实现幂等写入：在目标存储端基于聚合的key、timestamp等标识做幂等处理，确保即使重复发送数据也不会造成重复存储；
• 状态与写入绑定：将聚合状态标记为"待写入"状态，直到异步写入成功后再清空，故障恢复时可重新发送未确认的聚合数据。

内容的提问来源于stack exchange，提问作者Raúl García