{"record":{"id":"0e42b76bb19219dd","repo":"apache/flink","slug":"the-end-timestamp-of-a-processing-time-window-cann","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/OneInputWindowProcessOperator.java","lineNumber":265,"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":247,"sourceCodeEnd":283,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/OneInputWindowProcessOperator.java#L247-L283","documentation":"Thrown from the MergeFunction callback inside OneInputWindowProcessOperator while a record is added to a processing-time session window. After merging, the resulting window's maxTimestamp must still be later than the current processing time (mergeResult.maxTimestamp() > currentProcessingTime); a merged session that would already be due for trigger/cleanup is refused rather than created in an already-expired state. The message names the current processing time and the offending mergeResult window.","triggerScenarios":"WindowStrategy.session(gap, TimeType.PROCESSING): a record is assigned a session that merges with existing sessions into a window whose end (assignment time + gap) is <= current processing time at merge time. Happens when records spend longer than the session gap in queues (backpressure, slow source, GC pause, checkpoint stall) so wall-clock time overtakes the merged window's end.","commonSituations":"Session gap of a few seconds under backpressure or checkpoint stalls; bursty catch-up sources (file/Kafka after idle); overloaded TaskManagers where processing time outruns throughput; tests that pre-generate element timestamps and then replay them slower than real time.","solutions":["Increase the session gap so gap > worst-case per-record delay (queueing + backpressure + GC) - the merged session then never expires before the merge completes.","Fix the throughput problem: raise parallelism, reduce per-record work, check backpressure in the Web UI, enlarge network buffers.","Prefer event-time session windows (WindowStrategy.session(gap, TimeType.EVENT)) when correctness must not depend on wall-clock pacing.","In tests, drive processing-time windows with a controlled clock/pipelined input instead of sleeping between records."],"exampleFix":"// before: gap smaller than queueing delay\nstream.process(windowFn, WindowStrategy.session(Duration.ofSeconds(2), WindowStrategy.PROCESSING_TIME));\n// after: gap dominates worst-case end-to-end delay\nstream.process(windowFn, WindowStrategy.session(Duration.ofMinutes(1), WindowStrategy.PROCESSING_TIME));","handlingStrategy":"validation","validationCode":"// fail fast at job setup: processing-time sessions need gap > worst-case record delay\nlong gapMs = sessionGap.toMillis();\nif (gapMs <= measuredMaxRecordDelayMs) {\n    throw new IllegalArgumentException(\n            \"processing-time session gap (\" + gapMs + \"ms) must exceed max record delay (\"\n                    + measuredMaxRecordDelayMs + \"ms)\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Monitor backpressure; the session gap must dominate worst-case queueing delay","Prefer event-time sessions where wall-clock pacing affects correctness","Load-test processing-time session jobs at production throughput before deploying"],"tags":["flink","datastream-v2","windowing","session-windows","processing-time","backpressure"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}