apache/flink · error · UnsupportedOperationException

The end timestamp of an event-time window cannot become…

Error message

The end timestamp of an event-time window cannot become earlier than the current watermark by merging. Current event time: {eventTime} window: {mergeResult}

What it means

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.

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.

Example fix

// before: one input lags; its late records throw on merge
WatermarkStrategy.<L>forMonotonousTimestamps(); // left input
// after: bound out-of-orderness on every input
WatermarkStrategy.<L>forBoundedOutOfOrderness(Duration.ofMinutes(10));
WatermarkStrategy.<R>forBoundedOutOfOrderness(Duration.ofMinutes(10));

Defensive patterns

Strategy: validation

Validate before calling

// per-input guard before the two-input window operator
long wm = ctx.currentWatermark(); // min watermark of both inputs downstream
if (record.timestamp() + sessionGapMs + maxDelayMs <= wm) {
    ctx.output(LATE_TAG, record);
    return;
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/3e581626716fd41c. Report an issue: GitHub.

Appendix: source

Thrown at flink-datastream/src/main/java/org/apache/flink/datastream/impl/extension/window/operators/TwoInputNonBroadcastWindowProcessOperator.java:289

                // is the merged window and we work with that. If we don't merge then
                // actualWindow == window
                W actualWindow =
                        mergingWindows.addWindow(
                                window,
                                new MergingWindowSet.MergeFunction<>() {
                                    @Override
                                    public void merge(
                                            W mergeResult,
                                            Collection<W> mergedWindows,
                                            W stateWindowResult,
                                            Collection<W> mergedStateWindows)
                                            throws Exception {

                                        if ((windowAssigner.isEventTime()
                                                && mergeResult.maxTimestamp() + allowedLateness
                                                        <= internalTimerService
                                                                .currentWatermark())) {
                                            throw new UnsupportedOperationException(
                                                    "The end timestamp of an "
                                                            + "event-time window cannot become earlier than the current watermark "
                                                            + "by merging. Current event time: "
                                                            + internalTimerService
                                                                    .currentWatermark()
                                                            + " window: "
                                                            + mergeResult);
                                        } else if (!windowAssigner.isEventTime()) {
                                            long currentProcessingTime =
                                                    internalTimerService.currentProcessingTime();
                                            if (mergeResult.maxTimestamp()
                                                    <= currentProcessingTime) {
                                                throw new UnsupportedOperationException(
                                                        "The end timestamp of a "
                                                                + "processing-time window cannot become earlier than the current processing time "
                                                                + "by merging. Current processing time: "
                                                                + currentProcessingTime
                                                                + " window: "

View on GitHub (pinned to 2f3c205e92)