You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "MartijnVisser (via GitHub)" <gi...@apache.org> on 2026/08/28 13:07:15 UTC
[PR] [FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain [flink]
MartijnVisser opened a new pull request, #29037:
URL: https://github.com/apache/flink/pull/29037
## What is the purpose of the change
When a job is drained (bounded source finishing, or stop-with-savepoint --drain), `advanceToEndOfEventTime()` emits `Watermark.MAX_WATERMARK` directly into the operator chain output. However, `RecordWriterOutput#emitWatermark` and `ChainingOutput#emitWatermark` drop watermarks while the last announced `WatermarkStatus` is `IDLE`. A source whose last announced status was IDLE therefore never delivered the final MAX watermark, so downstream event-time timers and windows did not fire at end of job. This PR re-activates the output before emitting the final MAX watermark, the same pattern used by `TimestampsAndWatermarksOperator.WatermarkEmitter` and the table `WatermarkAssignerOperator`.
## Brief change log
- `SourceOperatorStreamTask`, `SourceStreamTask`, and `MultipleInputStreamTask` now emit `WatermarkStatus.ACTIVE` before `Watermark.MAX_WATERMARK` in `advanceToEndOfEventTime()`; downstream outputs deduplicate the ACTIVE when already active.
- The ACTIVE emission is skipped for tasks deployed as finished on restore: they never ran a source, their outputs were never idle, and `FinishedOnRestoreMainOperatorOutput` rejects watermark status events by design.
- Added `SourceOperatorStreamTaskIdleDrainTest` (task-level: a source that goes IDLE and then finishes must drain `[IDLE, ACTIVE, MAX_WATERMARK, EndOfData]`) and `ChainingOutputIdleMaxWatermarkTest` (output-level: ACTIVE-then-MAX after IDLE is delivered; MAX alone while idle is dropped, characterizing the gate the fix works around).
## Verifying this change
This change added tests and can be verified as follows:
- `SourceOperatorStreamTaskIdleDrainTest` fails without the fix (drained output was `[WatermarkStatus(IDLE), EndOfData{mode=DRAIN}]`, no MAX watermark) and passes with it.
- `ChainingOutputIdleMaxWatermarkTest` verifies the output-level contract.
- Regression: `SourceOperatorStreamTaskTest` (incl. `testEmittingMaxWatermarkAfterReadingAllRecords`), `SourceStreamTaskTest`, `OneInputStreamTaskTest`, and `MultipleInputStreamTaskTest` all pass locally, confirming the ACTIVE emission does not break the non-idle and finished-on-restore drain paths.
No existing tests were modified; both 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 — stop-with-savepoint --drain and end-of-input now emit a `WatermarkStatus.ACTIVE` before the final MAX watermark (deduplicated when already active); finished-on-restore tasks are exempted
- 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?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Fable 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #29037:
URL: https://github.com/apache/flink/pull/29037#issuecomment-5475993385
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain [flink]
Posted by "SEPURI-SAI-KRISHNA (via GitHub)" <gi...@apache.org>.
SEPURI-SAI-KRISHNA commented on PR #29037:
URL: https://github.com/apache/flink/pull/29037#issuecomment-5455504781
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` thro
ws `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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29037:
URL: https://github.com/apache/flink/pull/29037#issuecomment-5452937483
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fcbbddd6ffc9b324e1d2f839ffe813a40c427f4d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fcbbddd6ffc9b324e1d2f839ffe813a40c427f4d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fcbbddd6ffc9b324e1d2f839ffe813a40c427f4d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain [flink]
Posted by "SEPURI-SAI-KRISHNA (via GitHub)" <gi...@apache.org>.
SEPURI-SAI-KRISHNA commented on PR #29037:
URL: https://github.com/apache/flink/pull/29037#issuecomment-5477144118
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #29037:
URL: https://github.com/apache/flink/pull/29037#issuecomment-5475993058
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org