{"record":{"id":"e220bdd1beab6704","repo":"apache/flink","slug":"the-end-timestamp-of-a-processing-time-window-cann-e220bd","errorCode":null,"errorMessage":"The end timestamp of a processing-time window cannot become earlier than the current processing time by merging. Current processing time: {processingTime} window: {mergeResult}","messagePattern":"The end timestamp of a processing-time window cannot become earlier than the current processing time by merging\\. Current processing 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":302,"sourceCode":"\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: \"\n                                                                + mergeResult);\n                                            }\n                                        }\n\n                                        triggerContext.setKey(key);\n                                        triggerContext.setWindow(mergeResult);\n\n                                        triggerContext.onMerge(mergedWindows);\n\n                                        for (W m : mergedWindows) {\n                                            triggerContext.setWindow(m);\n                                            triggerContext.clear();\n                                            WindowUtils.deleteCleanupTimer(","sourceCodeStart":284,"sourceCodeEnd":320,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java#L284-L320","documentation":"Thrown from the MergeFunction callback in TwoInputNonBroadcastWindowProcessOperator while a left-input record is added to a processing-time session window. The merged window must satisfy mergeResult.maxTimestamp() > currentProcessingTime; a merged session already at or past the current wall clock is refused with UnsupportedOperationException because it would be born expired. The message names the current processing time and the offending mergeResult window.","triggerScenarios":"WindowStrategy.session(gap, TimeType.PROCESSING) on a two-input job: a left-input record merges sessions into a window whose end (arrival time + gap) is <= current processing time, which occurs when the record spent longer than the session gap in queues (backpressure, slow source, GC) before reaching the operator.","commonSituations":"Two-input jobs where one input is slow and backpressures the operator; small session gaps; checkpoint stalls; bursty catch-up after idle periods; overloaded TaskManagers.","solutions":["Increase the session gap beyond the worst-case queueing/backpressure delay observed for both inputs.","Diagnose and fix backpressure (Web UI backpressure tab, raise parallelism, reduce per-record work, check the slower input).","Switch to event-time session windows so merge validity depends on watermarks, not wall-clock pacing.","In tests, pump both inputs without artificial wall-clock delays, or use a controlled processing-time service."],"exampleFix":"// before\n... .process(twoInputFn, WindowStrategy.session(Duration.ofSeconds(2), WindowStrategy.PROCESSING_TIME));\n// after: gap dominates worst-case delay, or move to event time\n... .process(twoInputFn, WindowStrategy.session(Duration.ofMinutes(1), WindowStrategy.PROCESSING_TIME));","handlingStrategy":"validation","validationCode":"// setup-time assertion: gap must exceed worst-case delay of the slower input\nif (sessionGap.toMillis() <= maxObservedDelayMs) {\n    throw new IllegalArgumentException(\n            \"session gap too small for observed delay \" + maxObservedDelayMs + \"ms\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["The slower input of a two-input job dictates the minimum session gap","Fix backpressure on either input before tuning gaps","Prefer event-time sessions for two-input jobs in production"],"tags":["flink","datastream-v2","windowing","session-windows","two-input","processing-time","backpressure"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}