[FLINK-39586][connector-base] Add watermark stall reproduction ITCase

[MartijnVisser]

What is the purpose of the change

Reproduces FLINK-39586 end-to-end: after a HybridSource switches from a bounded to an unbounded source, a subtask that receives no splits from the unbounded source neither advances its watermark nor signals idleness. Its last bounded-era watermark then permanently caps the downstream combined watermark (CombinedWatermarkStatus.updateCombinedWatermark() short-circuits on the empty output set without updating the idle flag).

This is a reproduction PR and intentionally a draft: the new ITCase fails on master, so CI is expected to be red until a fix lands. The fix semantics (IDLE vs MAX_WATERMARK vs a dedicated FINISHED status, cf. FLINK-38477) are being discussed on the ticket and should be settled before a fix PR. The test only asserts user-visible behavior — the downstream watermark eventually passes the switch point — so it remains valid under any of those outcomes. A reference fix that makes it pass exists on branch FLINK-39586 of this fork.

Two observations from building the reproduction:

• The stall does not require checkpoint/restore, unlike the scenario in the ticket report. Any bounded-to-unbounded switch that leaves a subtask without splits triggers it; restore is only how the stale idle=false state arose in the reported environment.
• The mid-job case is why a finished-splits-emit-MAX_WATERMARK answer is not sufficient here: the unbounded successor continues the same timeline, and the subtask never reaches END_OF_INPUT, so the task-level MAX_WATERMARK path (advanceToEndOfEventTime, cf. FLINK-40499) never runs.

Brief change log

• Added HybridSourceWatermarkIdlenessITCase: a parallelism-2 HybridSource job where the bounded phase assigns one split per subtask (per-split watermarks are emitted before the splits finish), the unbounded successor assigns its single split to subtask 0 with an enumerator that never signals no-more-splits (like a Kafka reader owning no partition), and a parallelism-1 sink observes the combined downstream watermark.

Verifying this change

This change added tests and can be verified as follows:

• On master the ITCase fails: the record-count sanity check passes (the source switch completes and the unbounded split emits), then the watermark assertion times out with "Downstream watermark stalled at a bounded-era value".
• Cherry-picked onto the reference fix branch (FLINK-39586 on this fork), the ITCase passes.

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
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• 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)

🤖 Generated with Claude Code

[flinkbot]

CI report:

• 97a3b8d Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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