{"record":{"id":"928aa50d874cfb33","repo":"apache/flink","slug":"the-window-uses-a-merging-assigner-but-the-window-928aa5","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/TwoInputNonBroadcastWindowProcessOperator.java","lineNumber":199,"sourceCode":"\n        if (rightWindowStateDescriptor != null) {\n            rightWindowState =\n                    getOrCreateKeyedState(\n                            windowSerializer.createInstance(),\n                            windowSerializer,\n                            rightWindowStateDescriptor);\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 (leftWindowState instanceof InternalMergingState) {\n                leftWindowMergingState =\n                        (InternalMergingState<K, W, IN1, IN1, StateIterator<IN1>, Iterable<IN1>>)\n                                leftWindowState;\n            } else if (leftWindowState != null) {\n                throw new IllegalStateException(\n                        \"The window uses a merging assigner, but the window state is not mergeable.\");\n            }\n\n            if (rightWindowState instanceof InternalMergingState) {\n                rightWindowMergingState =\n                        (InternalMergingState<K, W, IN2, IN2, StateIterator<IN2>, Iterable<IN2>>)\n                                rightWindowState;\n            } else if (rightWindowState != 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});","sourceCodeStart":181,"sourceCodeEnd":217,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java#L181-L217","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Set state.backend to 'hashmap' or 'rocksdb'/'forst' whose list states implement InternalMergingState.","Implement InternalMergingState in the custom backend's list states, or avoid session windows on that backend.","Ensure the left/right window state descriptors register plain ListStateDescriptor and are not wrapped by job code.","Align flink-datastream-api and flink-dist versions on the classpath and cluster."],"exampleFix":"// before: two-input session window over a backend with non-mergeable state\nstream1.join(stream2).process(twoInputFn, WindowStrategy.session(Duration.ofMinutes(5)));\n// after: non-merging tumbling window, or a merging-capable backend\n... .process(twoInputFn, WindowStrategy.tumbling(Duration.ofMinutes(5), WindowStrategy.EVENT_TIME));","handlingStrategy":"validation","validationCode":"// in job setup, before execute()\nboolean merging = strategy instanceof SessionWindowStrategy;\nString backend = config.get(StateBackendOptions.STATE_BACKEND);\nif (merging && !Arrays.asList(\"hashmap\", \"rocksdb\", \"forst\").contains(backend)) {\n    throw new IllegalArgumentException(\n            \"Two-input session windows need a mergeable-state backend, got: \" + backend);\n}","typeGuard":"static boolean requiresMergingState(WindowStrategy s) {\n    return s instanceof SessionWindowStrategy;\n}","tryCatchPattern":null,"preventionTips":["For two-input session jobs, pin the backend explicitly in the job config","If a custom backend is used, assert its states implement InternalMergingState in backend unit tests","Prefer non-merging window types for two-input operators when session semantics are optional"],"tags":["flink","datastream-v2","windowing","session-windows","two-input","state-backend","operator-init"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}