[FLINK-40505][runtime] Prune failed subtasks from watermark alignment

[MartijnVisser]

What is the purpose of the change

Fixes FLINK-40505: SourceCoordinator never removed a subtask's last reported watermark from the watermark-alignment aggregation. After a subtask failure and reset, the stale entry kept constraining maxAllowedWatermark for the entire alignment group until the restarted subtask happened to report again — and indefinitely if it never did (e.g. after downscale). The retained value is not a conservative bound either: a restarted attempt can resume from an earlier checkpoint position than its last report.

Brief change log

• Added WatermarkAggregator#remove(key), returning the new aggregated watermark iff the aggregate changed.
• SourceCoordinator#subtaskReset now removes the failed subtask's watermark and, when the per-source aggregate changed, propagates it to the group-level aggregator in the coordinator store via the same update path used for reported watermarks (extracted as updateAggregatedWatermarkOfGroup).
• Removal is done in subtaskReset (per-subtask, fires only when no execution attempt is alive) rather than per-attempt executionAttemptFailed: watermark alignment is mutually exclusive with concurrent execution attempts (enforced in the constructor and in handleReportedWatermark), and global failover recreates the coordinator. Subtasks pick up the updated maxAllowedWatermark via the existing periodic announceCombinedWatermark.

Known follow-up (out of scope here): the group-level aggregator in the coordinator store never removes an operatorName entry, so a permanently stopped source in a multi-source alignment group is still never pruned.

Verifying this change

This change added tests and can be verified as follows:

• New test SourceCoordinatorAlignmentTest#testWatermarkAlignmentStatePrunedAfterSubtaskFailureAndReset fails without the fix (announced maxAllowedWatermark stayed 1100 from the stale entry; expected 1200 after the constraining subtask fails and resets) and passes with it.
• New characterization test testSubtaskThatNeverReportedReceivesNoAlignmentEvents documents that only subtasks that have reported receive alignment events.
• Regression: SourceCoordinatorAlignmentTest, SourceCoordinatorTest, and SourceCoordinatorContextTest all pass locally.

Both new tests are additive; no existing tests were modified.

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 (coordinator-side, failover-time only)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — changes JobManager-side SourceCoordinator behavior on subtask reset during partial failover
• 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:

• c3a7efa Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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