[flink] Support idle watermark handling in coordinator-commit

[ifndef-SleePy]

Purpose

Follow-up to PIP-30 (#8220) / PR #8401. Add idle watermark handling on the coordinator-commit path.

Key design

The coordinator-commit path does not go through Flink's StatusWatermarkValve, so WatermarkAligner reproduces valve semantics on the coordinator side. Idle status is reported by each writer subtask alongside the watermark, one pair per (subtask, checkpoint), over the existing OperatorEvent channel.

Watermark aggregation rules

• No idle subtasks — min over all subtasks. The normal steady-state case.
• Some idle subtasks — min over the active ones only. When a subtask turns from idle back to active, two sub-cases:
  • Its new watermark is already at or above the last emitted watermark → include it in the min immediately.
  • Its new watermark is still below the last emitted watermark → keep excluding it until it catches up, so the global watermark never goes backwards.
• All subtasks idle — emit a one-shot flush of the max across every subtask's last known watermark, then latch to IDLE so subsequent all-idle checkpoints do not re-flush.

State & compatibility

• Aligner state is in-memory only, rebuilt on global failover.
• CheckpointCommittablesSerializer bumps to v2 with a trailing idle bit; v1 payloads deserialize as idle = false.

Tests

• Unit tests covering the new/changed classes: WatermarkAlignerTest, CheckpointCommittablesSerializerTest, WriterCommittablesTest, CoordinatorCommittingRowDataStoreWriteOperatorTest, CommittingWriteOperatorCoordinatorTest.
• IT case CoordinatorCommitITCase.testIdleWatermarkParityAcrossCommitPaths — end-to-end parity: scripted idle sequences (all-active / partial-idle / all-idle) under both sink.coordinator-commit.enabled = true/false, snapshot watermarks must match.