如何在Checkpoint中存储TimestampAssigner的highestEventTime状态？

问题根源 #

Flink的TimestampAssigner和WatermarkStrategy都不属于算子范畴，它们是水位线生成流程中的轻量级组件，不参与Flink的状态管理体系，因此实现CheckpointingFunction不会触发状态快照与恢复逻辑。要持久化highestEventTime，必须将逻辑迁移到支持状态管理的ProcessFunction算子中。

具体实现步骤 #

1. 实现带状态的ProcessFunction
   用ValueState托管highestEventTime，Flink会自动将该状态纳入Checkpoint机制，重启后自动恢复：

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import java.time.Duration;

public class AdjustTimestampProcessFunction extends ProcessFunction<Model, Model> {
    private static final Duration ALLOWED_DELAY = Duration.ofHours(12);
    private ValueState<Long> highestEventTimeState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态，指定名称、序列化器和默认初始值
        ValueStateDescriptor<Long> stateDesc = new ValueStateDescriptor<>(
                "highestEventTime",
                LongSerializer.INSTANCE,
                0L
        );
        highestEventTimeState = getRuntimeContext().getState(stateDesc);
    }

    @Override
    public void processElement(Model element, Context ctx, Collector<Model> out) throws Exception {
        long currentHighest = highestEventTimeState.value();
        long timestamp = element.getTimeMs();

        // 执行原时间戳调整逻辑
        if (timestamp > (currentHighest - ALLOWED_DELAY.toMillis())) {
            timestamp = Math.max(timestamp, currentHighest);
            highestEventTimeState.update(timestamp);
        }

        // 更新元素时间戳并输出
        element.setTimeMs(timestamp);
        out.collect(element);
    }
}

2. 修改作业拓扑
   先通过基础的TimestampAssigner提取原始时间戳、配置水位线策略，再通过上述ProcessFunction调整时间戳：

stream.assignTimestampsAndWatermarks(WatermarkStrategy.<Model>forBoundedOutOfOrderness(ALLOWED_DELAY)
        // 仅提取原始时间戳，不做调整
        .withTimestampAssigner((element, recordTimestamp) -> element.getTimeMs()))
// 接入时间戳调整逻辑，同时维护持久化状态
.process(new AdjustTimestampProcessFunction())
// 后续处理逻辑...

关键说明 #

• ValueState是Flink的托管状态，会自动参与Checkpoint的快照与恢复，无需手动实现CheckpointingFunction。
• 迁移后，作业重启或重新部署时，highestEventTime会从最近的Checkpoint中恢复，保证时间戳调整逻辑的连续性。

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