[Flink] Fix dedicated streaming compaction after snapshot expiration

[lilei1128]

Purpose

Make Flink Dedicated Streaming Compact jobs self-recoverable when snapshot
expiration races with compaction source reads or checkpoint recovery.

A previous compaction may expire the snapshot or manifest files currently
being read by the source, causing OutOfRangeException or
FileNotFoundException. The same expired snapshot or pending split may also be
restored from a checkpoint after failover, causing the job to repeatedly retry
the invalid snapshot and remain stuck.

The fix rebuilds the compaction baseline from the latest valid snapshot,
clears expired source state, synchronizes rebase information with all
compaction writers, and resumes incremental compaction without manual
intervention.

The core idea is to treat snapshot expiration as a coordinated source and
sink recovery:

• Detect expired snapshots or expiration-related read failures in the
  dedicated compaction source.
• Clear expired pending splits and rebuild a complete baseline from the latest
  valid snapshot.
• Mark the baseline splits as rebase records and broadcast them to all
  compaction writers.
• Recreate writer state from the same valid snapshot and discard in-flight
  splits already covered by the rebase.
• Persist the rebase progress in checkpoint state so that subsequent recovery
  can continue from a valid baseline.

Tests

• Added snapshot expiration and writer restore tests.
• Added source and aligned enumerator tests for expired pending splits.
• Added compaction operator tests for rebase ordering, in-flight splits, and checkpoint restore.
• Added multi-table watermark and rebase state tests.
• Added partitioner tests for bucket routing and rebase broadcast.
• Added an end-to-end Flink IT case covering snapshot expiration, failover, source rebase, and continued compaction.
• All targeted tests passed: 38 tests, 0 failures, 0 errors.