[flink] Resolve per-partition bucket count in sink bucket shuffle

[Kaixuan-Duan]

Purpose

Closes #4310.

After ALTER TABLE SET ('bucket.num' = N) rescales a partitioned table, partitions created
before the rescale keep their original bucket counts while new partitions use the new one.
However, the primary-key-table Flink sink's pre-write bucket shuffle still sharded records
with the table-level numBuckets captured at job submission time (a constructor argument
of FlinkRowDataChannelComputer). The sharding denominator then mismatched the actual
per-partition layout, scattering the records of one bucket across multiple writer subtasks.
Each writer reports its own log end offset for the same bucket, so the WriterState ends up
with conflicting per-bucket offsets, and RecoveryOffsetManager.putMergedOffset throws
IllegalStateException on restore — the job can never recover from checkpoint/savepoint.

This PR restores the "one bucket, exactly one writer" invariant by making the bucket shuffle
resolve each partition's actual bucket count at runtime.

Brief change log

• Add PartitionBucketCountResolver: resolves a partition's authoritative bucket count from
  cluster metadata at runtime and caches it locally. A partition's bucket count is immutable
  once created, so cached entries are valid forever and need no invalidation. Cache misses
  are handled in three ways:
  • partition exists → authoritative count; a single listPartitionInfos call warms up
    every partition of the table;
  • partition not yet created (dynamic partition creation is triggered by the downstream
    writer, so blocking here would deadlock) → fall back to the current table-level count,
    which is exactly what the partition will be created with;
  • metadata RPC failure → bounded retries, then fail fast; no silent degradation.
• FlinkRowDataChannelComputer now uses the per-partition count for both the bucket id and
  the sharding-mode decision, replacing the stale table-level value.
• FlinkSink passes TablePath/Configuration to the channel computer at both call sites.

Tests

• PartitionBucketCountResolverTest : three-way miss handling, immutable caching,
  bounded-retry fail-fast — pure JVM with an injected fake metadata source.
• PartitionBucketCountResolverITCase : cold-start authoritative
  resolution (the pre-rescale partition keeps its count while the table-level value differs),
  fallback matching the count the partition is actually created with, and metadata-call
  counting across cache hits. Runs at parallelism 3 so both the pre- and post-rescale
  partitions exercise the combine-mode sharding formula.
• UndoRecoveryITCase#testPartitionedTableRecoveryAfterBucketRescale (end-to-end, three
  phases): write → stop-with-savepoint → rescale → a new submission restores the savepoint
  and writes to the pre-rescale partition → a second recovery writes to the post-rescale
  partition; asserts exact per-key sums at every phase. Before the fix, the second restore
  failed with conflicting per-bucket offsets.
• Mutation-verified: reverting the channel-computer change makes the ITCase fail with
  bucket 1 of partition 2024-01: channels [0, 2] (one bucket written by two subtasks —
  the exact failure this PR fixes); bypassing the resolver makes all 5 unit tests fail.
• Regression: the full fluss-flink-common sink suite and fluss-common pass.

Known limitations

• The fallback value is cached for the job's lifetime: if a rescale that shrinks the
  count lands between the fallback and the actual partition creation, that partition keeps
  sharding with the stale count until a restart (sub-second window; rescales are rare).
• Checkpoints produced before this fix with fragmented WriterState cannot be recovered and
  must be discarded.

API and Format

No public API or RPC/storage format changes.