{"record":{"id":"985f3b1881a9957f","repo":"apache/flink","slug":"the-window-uses-a-merging-assigner-but-the-window","errorCode":null,"errorMessage":"The window uses a merging assigner, but the window state is not mergeable.","messagePattern":"The window uses a merging assigner, but the window state is not mergeable\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"critical","filePath":"flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/OneInputWindowProcessOperator.java","lineNumber":166,"sourceCode":"        // NOTE - the state may be null in the case of the overriding evicting window operator\n        if (windowStateDescriptor != null) {\n            windowState =\n                    getOrCreateKeyedState(\n                            windowSerializer.createInstance(),\n                            windowSerializer,\n                            windowStateDescriptor);\n        }\n\n        // create the typed and helper states for merging windows\n        if (windowAssigner instanceof MergingWindowAssigner) {\n\n            // store a typed reference for the state of merging windows - sanity check\n            if (windowState instanceof InternalMergingState) {\n                windowMergingState =\n                        (InternalMergingState<K, W, IN, IN, StateIterator<IN>, Iterable<IN>>)\n                                windowState;\n            } else if (windowState != null) {\n                throw new IllegalStateException(\n                        \"The window uses a merging assigner, but the window state is not mergeable.\");\n            }\n\n            @SuppressWarnings(\"unchecked\")\n            final Class<Tuple2<W, W>> typedTuple = (Class<Tuple2<W, W>>) (Class<?>) Tuple2.class;\n\n            final TupleSerializer<Tuple2<W, W>> tupleSerializer =\n                    new TupleSerializer<>(\n                            typedTuple, new TypeSerializer[] {windowSerializer, windowSerializer});\n\n            final ListStateDescriptor<Tuple2<W, W>> mergingSetsStateDescriptor =\n                    new ListStateDescriptor<>(\"merging-window-set\", tupleSerializer);\n\n            // get the state that stores the merging sets\n            mergingSetsState =\n                    getOrCreateKeyedState(\n                            VoidNamespaceSerializer.INSTANCE.createInstance(),\n                            VoidNamespaceSerializer.INSTANCE,","sourceCodeStart":148,"sourceCodeEnd":184,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/OneInputWindowProcessOperator.java#L148-L184","documentation":"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.","triggerScenarios":"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.","commonSituations":"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().","solutions":["Check the effective state backend (state.backend config); switch to 'hashmap' or 'rocksdb'/'forst', whose list states implement InternalMergingState.","If a custom state backend is mandatory, implement InternalMergingState for the list states it returns, or stop using session windows with that backend.","Verify the operator receives a plain ListStateDescriptor for window contents and that nothing in the job replaces or wraps the registered state.","Align flink-datastream(-api) and flink-dist versions if a mixed-version classpath is possible."],"exampleFix":"// before: session window over a custom backend with non-mergeable list state\nstream.process(windowFn, WindowStrategy.session(Duration.ofMinutes(5)));\n// after: non-merging window type works on any backend,\n// or pin a backend that supports merging state\nstream.process(windowFn, WindowStrategy.tumbling(Duration.ofMinutes(5), WindowStrategy.EVENT_TIME));\n//   or: config.set(StateBackendOptions.STATE_BACKEND, \"rocksdb\");","handlingStrategy":"validation","validationCode":"// fail fast at job setup instead of at task startup\nboolean session = strategy instanceof SessionWindowStrategy;\nString backend = config.get(StateBackendOptions.STATE_BACKEND);\nif (session && !Arrays.asList(\"hashmap\", \"rocksdb\", \"forst\").contains(backend)) {\n    throw new IllegalArgumentException(\n            \"Session windows require a state backend with mergeable window state, got: \" + backend);\n}","typeGuard":"static boolean isMergingStrategy(WindowStrategy s) {\n    return s instanceof SessionWindowStrategy;\n}","tryCatchPattern":null,"preventionTips":["Prefer tumbling/sliding windows unless session semantics are required","Pin state.backend to hashmap/rocksdb/forst for session-window jobs","Validate strategy + backend combination in job setup so the failure is a clear setup error, not a runtime crash"],"tags":["flink","datastream-v2","windowing","session-windows","state-backend","operator-init"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}