[Bug]: Flink drops watermark holds for records awaiting a checkpoint

[bvolpato]

What happened?

The Flink runner can emit an output watermark past records still buffered for @RequiresStableInput. Revalidated against Apache Beam commit ffcbd7e464f8244bd623d23e7fa9fcc750e81bbb.

BufferingDoFnRunner.checkpointCompleted() clears the minimum buffered timestamp unconditionally at line 312, after releasing only the acknowledged buffers. The current buffer can still contain records received after that checkpoint's snapshot. DoFnOperator.notifyCheckpointComplete() immediately recomputes the output watermark using this now-missing hold.

For a streaming, non-keyed identity DoFn with @RequiresStableInput, exactly-once checkpointing, one concurrent checkpoint, and maxBundleSize=1:

1. Receive A@10, then snapshot checkpoint 1.
2. Receive B@20 into the new active buffer.
3. Receive input watermark 100. The buffered timestamps currently hold output watermark at 10.
4. Complete checkpoint 1. A is emitted, but B remains buffered. The hold is reset to Long.MAX_VALUE, allowing output watermark 100.
5. Snapshot and complete checkpoint 2. B is emitted at timestamp 20, after output watermark 100.

Actual output from the source-method reproduction:

[watermark:10, record:A@10, watermark:100, record:B@20]

Expected: retain B's hold at 20 (or an earlier conservative hold) until B is emitted. Advancing the watermark to 100 can cause downstream event-time windows to finalize early or discard otherwise on-time records as late. The sequence occurs during normal processing with one checkpoint in flight.

The bundleStarted guard does not prevent this: ordinary bundle completion sets it to false, while BufferingDoFnRunner.finishBundle() leaves these records buffered. A size of 1 makes the sequence deterministic; a count or timeout boundary can also finish the bundle with larger sizes.

To reproduce in the existing test suite, copy the operator/harness setup from DoFnOperatorTest.testExactlyOnceBuffering(), with these changes before constructing the operator:

options.setStreaming(true);
options.setMaxBundleSize(1L);
options.setCheckpointingInterval(1L);
options.setCheckpointingMode("EXACTLY_ONCE");
options.setNumConcurrentCheckpoints(1);

Use an identity @ProcessElement @RequiresStableInput method and omit the example's @StartBundle and @FinishBundle hooks. Replace its value-only input coder with a full coder so checkpoint buffering preserves the test timestamps:

WindowedValues.FullWindowedValueCoder<String> windowedValueCoder =
    WindowedValues.getFullCoder(StringUtf8Coder.of(), GlobalWindow.Coder.INSTANCE);

After testHarness.open(), the regression sequence is:

testHarness.processElement(new StreamRecord<>(
    WindowedValues.timestampedValueInGlobalWindow("A", new Instant(10))));
testHarness.snapshot(1L, 0L);
testHarness.processElement(new StreamRecord<>(
    WindowedValues.timestampedValueInGlobalWindow("B", new Instant(20))));
testHarness.processWatermark(new Watermark(100));
assertThat(doFnOperator.getCurrentOutputWatermark(), is(10L));

doFnOperator.notifyCheckpointComplete(1L);
org.junit.Assert.assertTrue(
    "B@20 is still buffered; output watermark must not pass it",
    doFnOperator.getCurrentOutputWatermark() <= 20L);

// With the preceding assertion omitted, inspect getOutput() after this
// checkpoint to observe B@20 arriving after watermark 100.
testHarness.snapshot(2L, 0L);
doFnOperator.notifyCheckpointComplete(2L);

Validation: compiled and ran a standalone Java harness extracting unchanged checkpoint, buffering, bundle-finishing, watermark, and checkpoint-completion methods from the commit above, with in-memory dependency stubs. It confirmed bundleStarted=false, B still buffered, and output watermark 100 after checkpoint 1. The proposed Flink JUnit regression above has not been run; this report does not claim a full Flink integration-test result.

Issue Priority

Priority: 1 (data loss / total loss of function)

Issue Components

• Component: Flink Runner
• Component: Java SDK