{"record":{"id":"7c2a88897e707295","repo":"apache/flink","slug":"the-window-uses-a-merging-assigner-but-the-window-7c2a88","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/TwoOutputWindowProcessOperator.java","lineNumber":175,"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":157,"sourceCodeEnd":193,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoOutputWindowProcessOperator.java#L157-L193","documentation":"Sanity check during TwoOutputWindowProcessOperator initialization (windows with a main and side output). When the assigner is a MergingWindowAssigner (session windows), the per-window state must implement InternalMergingState to merge contents of merged-away windows; a non-null windowState that fails the instanceof check makes open() throw IllegalStateException, so the job fails to start. Standard backends (hashmap, RocksDB/ForSt) satisfy this, pointing to a custom backend or state registration as the cause.","triggerScenarios":"A two-output windowed stream (e.g. WindowProcessFunction emitting to main + side collectors) built with WindowStrategy.session(...) while the keyed state for window contents is not an InternalMergingState - custom KeyedStateBackend, decorated state, or unsupported backend flavor.","commonSituations":"Pluggable state backends; jobs migrated from heap to a custom backend without checking merging support; mixed Flink versions; state factories wrapping returned states.","solutions":["Run the job on 'hashmap' or 'rocksdb'/'forst'.","Make the custom backend's list states implement InternalMergingState, or drop session windows for that backend.","Keep the window state descriptor a plain ListStateDescriptor end-to-end.","Align flink-datastream-api and flink-dist versions."],"exampleFix":"// before: session window, two-output function, non-merging backend\nstream.process(twoOutputFn, WindowStrategy.session(Duration.ofMinutes(5)));\n// after: non-merging strategy or merging-capable backend\nstream.process(twoOutputFn, WindowStrategy.tumbling(Duration.ofMinutes(5), WindowStrategy.EVENT_TIME));","handlingStrategy":"validation","validationCode":"// setup guard for two-output session jobs\nif (strategy instanceof SessionWindowStrategy\n        && !Arrays.asList(\"hashmap\", \"rocksdb\", \"forst\").contains(backendName)) {\n    throw new IllegalArgumentException(\"two-output session windows need a mergeable backend\");\n}","typeGuard":"static boolean requiresMergingState(WindowStrategy s) {\n    return s instanceof SessionWindowStrategy;\n}","tryCatchPattern":null,"preventionTips":["Pin the state backend explicitly for session-window jobs","Prefer non-merging strategies for two-output windows when possible","Assert strategy/backend compatibility in job setup so failures are clear and early"],"tags":["flink","datastream-v2","windowing","session-windows","two-output","state-backend","operator-init"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}