Flink实例（129）：状态管理(十八)Table API 和 SQL 模块状态管理(三) Flink SQL空闲状态保留时间（idle state retention time）实现原理

stenv.getConfig().setIdleStateRetentionTime(Time.hours(24), Time.hours(36))

public interface CleanupState {
    default void registerProcessingCleanupTimer(
            ValueState<Long> cleanupTimeState,
            long currentTime,
            long minRetentionTime,
            long maxRetentionTime,
            TimerService timerService)
            throws Exception {
        // last registered timer
        Long curCleanupTime = cleanupTimeState.value();
 
        // check if a cleanup timer is registered and
        // that the current cleanup timer won‘t delete state we need to keep
        if (curCleanupTime == null || (currentTime + minRetentionTime) > curCleanupTime) {
            // we need to register a new (later) timer
            long cleanupTime = currentTime + maxRetentionTime;
            // register timer and remember clean-up time
            timerService.registerProcessingTimeTimer(cleanupTime);
            // delete expired timer
            if (curCleanupTime != null) {
                timerService.deleteProcessingTimeTimer(curCleanupTime);
            }
            cleanupTimeState.update(cleanupTime);
        }
    }
}

public abstract class KeyedProcessFunctionWithCleanupState<K, IN, OUT>
        extends KeyedProcessFunction<K, IN, OUT> implements CleanupState {
    private static final long serialVersionUID = 2084560869233898457L;
 
    private final long minRetentionTime;
    private final long maxRetentionTime;
    protected final boolean stateCleaningEnabled;
 
    // holds the latest registered cleanup timer
    private ValueState<Long> cleanupTimeState;
 
    public KeyedProcessFunctionWithCleanupState(long minRetentionTime, long maxRetentionTime) {
        this.minRetentionTime = minRetentionTime;
        this.maxRetentionTime = maxRetentionTime;
        this.stateCleaningEnabled = minRetentionTime > 1;
    }
 
    protected void initCleanupTimeState(String stateName) {
        if (stateCleaningEnabled) {
            ValueStateDescriptor<Long> inputCntDescriptor =
                    new ValueStateDescriptor<>(stateName, Types.LONG);
            cleanupTimeState = getRuntimeContext().getState(inputCntDescriptor);
        }
    }
 
    protected void registerProcessingCleanupTimer(Context ctx, long currentTime) throws Exception {
        if (stateCleaningEnabled) {
            registerProcessingCleanupTimer(
                    cleanupTimeState,
                    currentTime,
                    minRetentionTime,
                    maxRetentionTime,
                    ctx.timerService());
        }
    }
 
    protected boolean isProcessingTimeTimer(OnTimerContext ctx) {
        return ctx.timeDomain() == TimeDomain.PROCESSING_TIME;
    }
 
    protected void cleanupState(State... states) {
        for (State state : states) {
            state.clear();
        }
        this.cleanupTimeState.clear();
    }
 
    protected Boolean needToCleanupState(Long timestamp) throws IOException {
        if (stateCleaningEnabled) {
            Long cleanupTime = cleanupTimeState.value();
            // check that the triggered timer is the last registered processing time timer.
            return timestamp.equals(cleanupTime);
        } else {
            return false;
        }
    }
}

@Override
public void processElement(RowData input, Context context, Collector<RowData> out) throws Exception {
    long currentTime = context.timerService().currentProcessingTime();
    // register state-cleanup timer
    registerProcessingCleanupTimer(context, currentTime);
    // ......
}

@Override
public void onTimer(
        long timestamp,
        OnTimerContext ctx,
        Collector<RowData> out) throws Exception {
    if (stateCleaningEnabled) {
        // cleanup cache
        kvSortedMap.remove(keyContext.getCurrentKey());
        cleanupState(dataState);
    }
}