apache/flink · critical · IllegalStateException

The window uses a merging assigner, but the window state is

Error message

The window uses a merging assigner, but the window state is not mergeable.

What it means

Sanity check during OneInputWindowProcessOperator initialization. When the WindowAssigner is a MergingWindowAssigner (i.e. a session window strategy), merged-away windows' contents must be merged in state, so the registered per-window state must implement InternalMergingState. The operator casts the state obtained from windowStateDescriptor; if it is non-null but not an InternalMergingState, it throws IllegalStateException at open() and the job never runs. With the built-in backends (heap/hashmap, RocksDB/ForSt) list states implement InternalMergingState, so this signals a non-standard state backend or state registration.

Source

Thrown at flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/OneInputWindowProcessOperator.java:166

        // NOTE - the state may be null in the case of the overriding evicting window operator
        if (windowStateDescriptor != null) {
            windowState =
                    getOrCreateKeyedState(
                            windowSerializer.createInstance(),
                            windowSerializer,
                            windowStateDescriptor);
        }

        // create the typed and helper states for merging windows
        if (windowAssigner instanceof MergingWindowAssigner) {

            // store a typed reference for the state of merging windows - sanity check
            if (windowState instanceof InternalMergingState) {
                windowMergingState =
                        (InternalMergingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>)
                                windowState;
            } else if (windowState != null) {
                throw new IllegalStateException(
                        "The window uses a merging assigner, but the window state is not mergeable.");
            }

            @SuppressWarnings("unchecked")
            final Class<Tuple2<W, W>> typedTuple = (Class<Tuple2<W, W>>) (Class<?>) Tuple2.class;

            final TupleSerializer<Tuple2<W, W>> tupleSerializer =
                    new TupleSerializer<>(
                            typedTuple, new TypeSerializer[] {windowSerializer, windowSerializer});

            final ListStateDescriptor<Tuple2<W, W>> mergingSetsStateDescriptor =
                    new ListStateDescriptor<>("merging-window-set", tupleSerializer);

            // get the state that stores the merging sets
            mergingSetsState =
                    getOrCreateKeyedState(
                            VoidNamespaceSerializer.INSTANCE.createInstance(),
                            VoidNamespaceSerializer.INSTANCE,

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Check the effective state backend (state.backend config); switch to 'hashmap' or 'rocksdb'/'forst', whose list states implement InternalMergingState.
2. If a custom state backend is mandatory, implement InternalMergingState for the list states it returns, or stop using session windows with that backend.
3. Verify the operator receives a plain ListStateDescriptor for window contents and that nothing in the job replaces or wraps the registered state.
4. Align flink-datastream(-api) and flink-dist versions if a mixed-version classpath is possible.

Example fix

// before: session window over a custom backend with non-mergeable list state
stream.process(windowFn, WindowStrategy.session(Duration.ofMinutes(5)));
// after: non-merging window type works on any backend,
// or pin a backend that supports merging state
stream.process(windowFn, WindowStrategy.tumbling(Duration.ofMinutes(5), WindowStrategy.EVENT_TIME));
//   or: config.set(StateBackendOptions.STATE_BACKEND, "rocksdb");

// fail fast at job setup instead of at task startup
boolean session = strategy instanceof SessionWindowStrategy;
String backend = config.get(StateBackendOptions.STATE_BACKEND);
if (session && !Arrays.asList("hashmap", "rocksdb", "forst").contains(backend)) {
    throw new IllegalArgumentException(
            "Session windows require a state backend with mergeable window state, got: " + backend);
}

static boolean isMergingStrategy(WindowStrategy s) {
    return s instanceof SessionWindowStrategy;
}

When it happens

Trigger: Building a one-input windowed stream with WindowStrategy.session(...) (mapped to EventTimeSessionWindows or ProcessingTimeSessionWindows, both MergingWindowAssigners) while the keyed state returned for the window contents is not an InternalMergingState - typically a custom/third-party KeyedStateBackend, or a state factory that wraps or replaces the list state with a non-merging implementation.

Common situations: Custom state backend plugged into the cluster; wrapping the window state descriptor so the backend returns a plain state; mixed Flink versions where the window extension and backend disagree; running session windows on a backend flavor that never implemented merge().

Related errors

• The window uses a merging assigner, but the window state is
• The window uses a merging assigner, but the window state is
• Window {window} is not in in-flight window set.
• The end timestamp of an event-time window cannot become earl
• The end timestamp of a processing-time window cannot become

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/985f3b1881a9957f. Report an issue: GitHub.