You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@flink.apache.org by "kotwal-itpro (via GitHub)" <gi...@apache.org> on 2026/10/06 06:15:08 UTC

[PR] [FLINK-40501][datastream] Keep watermark status ordered with queued watermarks in AsyncWaitOperator [flink]

kotwal-itpro opened a new pull request, #29397:
URL: https://github.com/apache/flink/pull/29397

## What is the purpose of the change

`AsyncWaitOperator` puts watermarks into its async queue so they are emitted after the records that arrived before them, but it does not override `processWatermarkStatus`. `AbstractStreamOperator#processWatermarkStatus` forwards the status right away, so a `WatermarkStatus` can overtake watermarks that are still queued. For the input `[record, W100, IDLE]` the operator emits `[IDLE, record, W100]`.

A `StatusWatermarkValve` downstream ignores watermarks that arrive on an idle channel, so W100 is lost, not just delayed. If it was the last watermark on that channel (e.g. `MAX_WATERMARK` at drain), downstream event time never advances past it.

## Brief change log

- Added `WatermarkStatusQueueEntry`, a queue entry that is always completed and emits a `WatermarkStatus`.
- `OrderedStreamElementQueue` and `UnorderedStreamElementQueue` accept watermark statuses. The unordered queue gives a status its own segment, like a watermark, so records cannot be reordered across it.
- `AsyncWaitOperator#processWatermarkStatus` adds the status to the queue and emits completed elements, the same way as `processWatermark`. A status with an empty queue is still emitted immediately.
- Statuses that are still queued at checkpoint time are part of the snapshotted queue elements, and `open()` replays them on restore.

## Verifying this change

This change added tests and can be verified as follows:

- `AsyncWaitOperatorTest#testWatermarkStatusDoesNotOvertakeWatermark{Ordered,Unordered}`: `[record, W100, IDLE]` with the async result withheld emits nothing until the record completes, then `[record, W100, IDLE]`.
- `AsyncWaitOperatorTest#testWatermarkStatusIsRestoredInOrder`: a status that is queued at snapshot time is restored and emitted after the restored record and watermark.
- `AsyncWaitOperatorTest#testWatermarkStatusWithEmptyQueueIsEmittedImmediately`: no extra latency when nothing is queued.
- `StreamElementQueueTest#testWatermarkStatusIsEmittedInOrder` (both queues) and `UnorderedStreamElementQueueTest#testWatermarkStatusCompletionOrder`.

All new tests except the empty-queue one fail without the change. `AsyncWaitOperatorTest`, `StreamElementQueueTest`, `OrderedStreamElementQueueTest` and `UnorderedStreamElementQueueTest` pass.

## 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 operator state already uses `StreamElementSerializer`, which supports `WatermarkStatus`)
- The runtime per-record code paths (performance sensitive): no (only the watermark status path)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes. The async operator state can now contain `WatermarkStatus` elements. Restoring such a checkpoint with a version without this change fails in `AsyncWaitOperator#open()` with "Unknown record type".
- 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

`RowTimeMiniBatchAssignerOperator` has the same shape (it buffers watermarks but forwards the status immediately), as noted in the JIRA. I kept this PR to `AsyncWaitOperator` and can follow up on that one separately.

---

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

- [X] Yes (please specify the tool below)

Generated-by: Claude Opus 5.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-40501][datastream] Keep watermark status ordered with queued watermarks in AsyncWaitOperator [flink]

Posted by "flinkbot (via GitHub)" <gi...@apache.org>.

flinkbot commented on PR #29397:
URL: https://github.com/apache/flink/pull/29397#issuecomment-6010662637

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "f7209ebe5c8d87ff433e7d83a0b790ce1bf262a5",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "f7209ebe5c8d87ff433e7d83a0b790ce1bf262a5",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * f7209ebe5c8d87ff433e7d83a0b790ce1bf262a5 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-40501][datastream] Keep watermark status ordered with queued watermarks in AsyncWaitOperator [flink]

Posted by "kotwal-itpro (via GitHub)" <gi...@apache.org>.

kotwal-itpro commented on PR #29397:
URL: https://github.com/apache/flink/pull/29397#issuecomment-6300854348

@snuyanzin, could you take a look at this one when you have time? You worked on `AsyncWaitOperator` recently (FLINK-40456), and this fixes another way it loses data: an IDLE watermark status overtakes a queued watermark, so downstream drops that watermark and event time can stop advancing. The second commit fixes the same pattern in `RowTimeMiniBatchAssignerOperator`.

I also have three smaller PRs open since Oct 6, all still merging cleanly, in case you or someone else has time for them:

- #29394 FLINK-40922 (planner): `SELECT DISTINCT` over a `GROUP BY ROLLUP` returns duplicate rows when a group key is NULL.
- #29395 FLINK-40919 (planner): `AVG` on an INT column overflows when it is rewritten to `SUM / COUNT`. This one is a draft because the fix has a planning trade-off I'd like a maintainer's view on.
- #29396 FLINK-40836 (checkpointing): restoring with `execution.state-recovery.without-channel-state.checkpoint-id` drops the coordinator state, so FLIP-27 sources re-read data they already emitted.

Happy to rebase or split anything if that helps.

--
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-40501][datastream] Keep watermark status ordered with queued watermarks in AsyncWaitOperator [flink]

Posted by "kotwal-itpro (via GitHub) via issues" <is...@flink.apache.org>.

kotwal-itpro commented on PR #29397:
URL: https://github.com/apache/flink/pull/29397#issuecomment-6596529115

   @snuyanzin, a gentle follow-up on this and #29394 and #29396, which are still waiting for a first review. All three still merge cleanly into master. If someone else is a better fit for the planner or checkpointing parts, I'm glad to ping them instead. Thanks!
   


-- 
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