Understanding Idle State Retention Time in Flink SQL

When using Flink SQL, a common mistake for beginners is forgetting to set the idle state retention time, which can cause state explosion as intermediate state accumulates for each grouping key.

Why Set Idle State Retention Time

Grouped processing stores intermediate results as state, and as the number of keys grows, the state size expands. Most of this state is time‑sensitive and does not need to be kept forever—for example, duplicate records in a Top‑N operation usually appear within a limited interval (e.g., one hour or one day). Flink SQL provides an idle state retention time feature that automatically clears state for keys that have not been updated for a configured period. The setting is applied as follows:

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

The setIdleStateRetentionTime() method requires two arguments: a minimum retention time ( minRetentionTime) and a maximum retention time ( maxRetentionTime), which must differ by at least five minutes.

How It Is Implemented

At the low level, idle state retention is represented by the o.a.f.table.runtime.functions.CleanupState interface:

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);
        }
    }
}

Each key maintains its own latest cleanup timestamp in a ValueState. A new timer is registered when either the ValueState is empty (first occurrence of the key) or the current time plus minRetentionTime exceeds the previously scheduled cleanup time. The timer is set to currentTime + maxRetentionTime, and any outdated timer is removed. If the interval between minRetentionTime and maxRetentionTime is too short, timers will be created and updated frequently, increasing overhead, so a longer interval is generally recommended.

The inheritance diagram of the CleanupState interface (shown in the image below) reveals that most functions supporting idle‑state cleanup inherit from the abstract class KeyedProcessFunctionWithCleanupState:

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;
        }
    }
}

The idle‑state retention currently works only with processing‑time semantics, and minRetentionTime must be greater than zero to take effect.

Concrete implementations use the helper methods provided by KeyedProcessFunctionWithCleanupState. For example, the AppendOnlyTopNFunction registers a cleanup timer in its processElement() method:

@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);
    // ......
}

When the timer fires, the onTimer() method invokes the base class’s cleanupState() to actually remove the stale state:

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

Beyond these functions, the Table/SQL module also includes a built‑in trigger StateCleaningCountTrigger that can clean windows based on element count or idle‑state retention thresholds (FIRE_AND_PURGE).