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 TwoInputNonBroadcastWindowProcessOperator initialization, applied to the left (input-1) window state. When the WindowAssigner is a MergingWindowAssigner (session windows), both inputs' window states must implement InternalMergingState so contents of merged-away windows can be merged; if leftWindowState is non-null but not mergeable, open() fails with IllegalStateException and the job never runs. Built-in backends (hashmap, RocksDB/ForSt) satisfy this, so the throw indicates a non-standard state backend or state registration.

Source

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

        if (rightWindowStateDescriptor != null) {
            rightWindowState =
                    getOrCreateKeyedState(
                            windowSerializer.createInstance(),
                            windowSerializer,
                            rightWindowStateDescriptor);
        }

        // 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 (leftWindowState instanceof InternalMergingState) {
                leftWindowMergingState =
                        (InternalMergingState<K, W, IN1, IN1, StateIterator<IN1>, Iterable<IN1>>)
                                leftWindowState;
            } else if (leftWindowState != null) {
                throw new IllegalStateException(
                        "The window uses a merging assigner, but the window state is not mergeable.");
            }

            if (rightWindowState instanceof InternalMergingState) {
                rightWindowMergingState =
                        (InternalMergingState<K, W, IN2, IN2, StateIterator<IN2>, Iterable<IN2>>)
                                rightWindowState;
            } else if (rightWindowState != 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});

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Set state.backend to 'hashmap' or 'rocksdb'/'forst' whose list states implement InternalMergingState.
2. Implement InternalMergingState in the custom backend's list states, or avoid session windows on that backend.
3. Ensure the left/right window state descriptors register plain ListStateDescriptor and are not wrapped by job code.
4. Align flink-datastream-api and flink-dist versions on the classpath and cluster.

Example fix

// before: two-input session window over a backend with non-mergeable state
stream1.join(stream2).process(twoInputFn, WindowStrategy.session(Duration.ofMinutes(5)));
// after: non-merging tumbling window, or a merging-capable backend
... .process(twoInputFn, WindowStrategy.tumbling(Duration.ofMinutes(5), WindowStrategy.EVENT_TIME));

// in job setup, before execute()
boolean merging = strategy instanceof SessionWindowStrategy;
String backend = config.get(StateBackendOptions.STATE_BACKEND);
if (merging && !Arrays.asList("hashmap", "rocksdb", "forst").contains(backend)) {
    throw new IllegalArgumentException(
            "Two-input session windows need a mergeable-state backend, got: " + backend);
}

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

When it happens

Trigger: A two-input windowed operator (e.g. interval/window join style job) built with WindowStrategy.session(...) where the keyed state returned for the left input's window contents does not implement InternalMergingState - custom KeyedStateBackend, wrapped/replaced list state, or a backend flavor without merge() support.

Common situations: Custom or third-party state backends; state factories that decorate the returned state; mixed Flink versions between API and runtime; session-window two-input jobs moved onto an unsupported backend.

Related errors

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

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