{"record":{"id":"3e581626716fd41c","repo":"apache/flink","slug":"the-end-timestamp-of-an-event-time-window-cannot-b-3e5816","errorCode":null,"errorMessage":"The end timestamp of an event-time window cannot become earlier than the current watermark by merging. Current event time: {eventTime} window: {mergeResult}","messagePattern":"The end timestamp of an event-time window cannot become earlier than the current watermark by merging\\. Current event time: (.+?) window: (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java","lineNumber":289,"sourceCode":"                // is the merged window and we work with that. If we don't merge then\n                // actualWindow == window\n                W actualWindow =\n                        mergingWindows.addWindow(\n                                window,\n                                new MergingWindowSet.MergeFunction<>() {\n                                    @Override\n                                    public void merge(\n                                            W mergeResult,\n                                            Collection<W> mergedWindows,\n                                            W stateWindowResult,\n                                            Collection<W> mergedStateWindows)\n                                            throws Exception {\n\n                                        if ((windowAssigner.isEventTime()\n                                                && mergeResult.maxTimestamp() + allowedLateness\n                                                        <= internalTimerService\n                                                                .currentWatermark())) {\n                                            throw new UnsupportedOperationException(\n                                                    \"The end timestamp of an \"\n                                                            + \"event-time window cannot become earlier than the current watermark \"\n                                                            + \"by merging. Current event time: \"\n                                                            + internalTimerService\n                                                                    .currentWatermark()\n                                                            + \" window: \"\n                                                            + mergeResult);\n                                        } else if (!windowAssigner.isEventTime()) {\n                                            long currentProcessingTime =\n                                                    internalTimerService.currentProcessingTime();\n                                            if (mergeResult.maxTimestamp()\n                                                    <= currentProcessingTime) {\n                                                throw new UnsupportedOperationException(\n                                                        \"The end timestamp of a \"\n                                                                + \"processing-time window cannot become earlier than the current processing time \"\n                                                                + \"by merging. Current processing time: \"\n                                                                + currentProcessingTime\n                                                                + \" window: \"","sourceCodeStart":271,"sourceCodeEnd":307,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java#L271-L307","documentation":"Thrown from the MergeFunction callback in TwoInputNonBroadcastWindowProcessOperator while a left-input (onRecord1) record is added to an event-time session window. A merge is only legal if the resulting window's maxTimestamp plus allowedLateness is still later than the current watermark; a merged session that would already be expired is refused with UnsupportedOperationException instead of touching possibly-cleaned state. The message reports the watermark and the offending mergeResult window.","triggerScenarios":"A late left-input record arrives for an event-time session window when watermark W already satisfies mergeResult.maxTimestamp() + allowedLateness <= W - the record would merge sessions into a window whose end is at/before the watermark. The session() factory exposes no allowedLateness, so effectively any expired merge result throws.","commonSituations":"Watermark out-of-orderness budget smaller than the real delay of either input stream; one input (e.g. the slower side of a join) delivering stale events; replay/backfill after watermarks advanced; per-input watermark asymmetry in two-input jobs.","solutions":["Size WatermarkStrategy.forBoundedOutOfOrderness on BOTH inputs to at least the max expected event delay plus the session gap.","Filter or side-output late records upstream of the windowed two-input operator using currentWatermark() comparisons.","Use a non-merging strategy with explicit lateness for the two-input window, e.g. WindowStrategy.tumbling(size, TimeType.EVENT, allowedLateness).","If session semantics are mandatory with late merges, use the DataStream v1 WindowOperator path until the session strategy supports allowedLateness."],"exampleFix":"// before: one input lags; its late records throw on merge\nWatermarkStrategy.<L>forMonotonousTimestamps(); // left input\n// after: bound out-of-orderness on every input\nWatermarkStrategy.<L>forBoundedOutOfOrderness(Duration.ofMinutes(10));\nWatermarkStrategy.<R>forBoundedOutOfOrderness(Duration.ofMinutes(10));","handlingStrategy":"validation","validationCode":"// per-input guard before the two-input window operator\nlong wm = ctx.currentWatermark(); // min watermark of both inputs downstream\nif (record.timestamp() + sessionGapMs + maxDelayMs <= wm) {\n    ctx.output(LATE_TAG, record);\n    return;\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Bound out-of-orderness on BOTH inputs; the operator watermark is the minimum of the two","Side-output late records before windowed two-input operators","Use non-merging strategies with explicit allowedLateness when session semantics are optional","Watch per-input watermark lag metrics to catch the slow side early"],"tags":["flink","datastream-v2","windowing","session-windows","two-input","event-time","late-data"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}