[FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain

[MartijnVisser]

What is the purpose of the change

At drain, advanceToEndOfEventTime() emits Watermark.MAX_WATERMARK into the chain output, but RecordWriterOutput and ChainingOutput drop watermarks while the announced status is IDLE. A source that was idle at drain therefore never delivers the final MAX watermark, and downstream event-time timers/windows don't fire. Fix: emit WatermarkStatus.ACTIVE first, the same pattern as TimestampsAndWatermarksOperator.WatermarkEmitter and WatermarkAssignerOperator.

Brief change log

• SourceOperatorStreamTask, SourceStreamTask, MultipleInputStreamTask: emit ACTIVE before MAX_WATERMARK in advanceToEndOfEventTime(); deduplicated downstream when already active.
• The first two skip the ACTIVE for tasks deployed as finished (FinishedOnRestoreMainOperatorOutput rejects status events). MultipleInputStreamTask needs no guard: its chained source outputs start ACTIVE, so the redundant status is deduplicated before reaching FinishedOnRestoreInput.
• Regression tests: SourceOperatorStreamTaskIdleDrainTest and MultipleInputStreamTaskIdleDrainTest (both red without the fix).
• ChainingOutputIdleMaxWatermarkTest is characterization only: documents the idle gate and the deduplication contract, passes with and without the fix.

Verifying this change

• SourceOperatorStreamTaskIdleDrainTest: red without the fix (drained output [IDLE, EndOfData], no MAX), green with it.
• MultipleInputStreamTaskIdleDrainTest: red with the MultipleInputStreamTask hunk reverted, green with it.
• Regression suites green: SourceOperatorStreamTaskTest, SourceStreamTaskTest, OneInputStreamTaskTest, TwoInputStreamTaskTest, MultipleInputStreamTaskTest, MultipleInputStreamTaskChainedSourcesCheckpointingTest (incl. the finished-on-restore + chained sources drain path).

No existing tests modified; all three test classes are new.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no (drain-time only)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — drain now emits ACTIVE before the final MAX watermark (deduplicated when already active)
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Fable 5)

[flinkbot]

CI report:

• be2272f Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[SEPURI-SAI-KRISHNA]

Not a reviewer on this, just read it closely because it touches a path I've been in. The change looks right to me, I traced the drain path through the valve as it stands on this base (so including #29024): ACTIVE reactivates the subpartition, realigning it immediately if its stored watermark is still at or above lastOutputWatermark and otherwise leaving MAX_WATERMARK to realign it, and either way findAndOutputNewMinWatermarkAcrossAlignedSubpartitions emits MAX as soon as the first source subtask drains. Two smaller things:

The isTaskDeployedAsFinished() guard in MultipleInputStreamTask appears to be dead code. It's clearly load-bearing in the other two tasks, where FinishedOnRestoreMainOperatorOutput#emitWatermarkStatus throws unconditionally. But getChainedSourceOutputs() returns ChainingOutput/CopyingChainingOutput (OperatorChain#createChainedSourceOutput), whose announcedStatus is initialised to ACTIVE and is never set to IDLE on a finished-on-restore chain, so emitWatermarkStatus(ACTIVE) is deduplicated inside the output and never reaches the input underneath. The comment also says the outputs reject status events, which holds for the other two tasks but not here: it's FinishedOnRestoreInput#processWatermarkStatus that would throw, and the dedup keeps it unreachable. I checked this with a scratch test against unmodified master: emitWatermarkStatus(ACTIVE) on a ChainingOutput wrapping a FinishedOnRestoreInput does not throw, while the same call with IDLE throws ExceptionInChainedOperatorException / IllegalStateException, so the input is genuinely reachable, and it is the dedup rather than the guard that keeps ACTIVE safe. CopyingChainingOutput behaves identically. Harmless either way, but the copied comment makes it read as necessary.

Which of the two new test classes is the regression test? ChainingOutput isn't touched by this PR, and I ran both tests from ChainingOutputIdleMaxWatermarkTest against unmodified master, both green, so they pass with and without the fix, useful as characterisation of the idle gate, but they don't guard the change. SourceOperatorStreamTaskIdleDrainTest is the one that does, I confirmed the red/green locally: without the fix it reports [WatermarkStatus(IDLE), EndOfData{mode=DRAIN}], with it the assertion passes. (Applying all three hunks, SourceOperatorStreamTaskTest, SourceStreamTaskTest, MultipleInputStreamTaskTest, OneInputStreamTaskTest and TwoInputStreamTaskTest are 102/102 green here too, for whatever that's worth alongside CI.) Might be worth saying so in the description, since "verifies the output-level contract" reads like it guards the fix.

Minor while you're in there: SourceOperatorStreamTaskTest already has testEmittingMaxWatermarkAfterReadingAllRecords, which is the exact non-idle counterpart of the new case on the same harness, the two would read well next to each other.

[MartijnVisser]

Updated fcbbddd → be2272f: dropped the dead guard in MultipleInputStreamTask and fixed the comment, added MultipleInputStreamTaskIdleDrainTest since the multi-input change had no test coverage of its own, and added a dedup test to ChainingOutputIdleMaxWatermarkTest. No existing tests modified.

[MartijnVisser]

@SEPURI-SAI-KRISHNA All three points check out. I've dropped the guard in MultipleInputStreamTask, fixed the comment, and marked ChainingOutputIdleMaxWatermarkTest as characterization in the description. Your first point also surfaced that the multi-input change had no test coverage of its own, so I've added MultipleInputStreamTaskIdleDrainTest. I'm keeping the drain tests in their own classes so no existing test files are touched.

[SEPURI-SAI-KRISHNA]

Thanks for picking these up. Checked the update: MultipleInputStreamTaskIdleDrainTest is genuinely red/green here, without the MultipleInputStreamTask hunk it fails with actual: [WatermarkStatus(IDLE)], with it green.

Since dropping the guard was my suggestion I also ran the finished-on-restore paths against be2272f: MultipleInputStreamTaskTest, MultipleInputStreamTaskChainedSourcesCheckpointingTest, StreamTaskFinalCheckpointsTest, SourceOperatorStreamTaskTest, SourceStreamTaskTest, OneInputStreamTaskTest, TwoInputStreamTaskTest and the new test, 136/136 green, plus 3/3 in ChainingOutputIdleMaxWatermarkTest. Worth noting for the record that FinishedOperatorChain goes through the normal OperatorChain constructor, so chainedSources is populated and the loop really does run on a task deployed as finished, the dedup is what makes the unconditional ACTIVE safe, as your new comment says.

LGTM, non-binding.