[flink] Bucket shuffle uses a fixed table-level numBuckets, breaking sink recovery after partition rescale

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Fluss version

main (development)

### Please describe the bug 🐞

**Background**: PR #3908 introduced per-partition bucket counts, allowing partitioned tables to
be rescaled via `ALTER TABLE SET ('bucket.num' = xx)`: existing partitions keep their original
bucket counts, while partitions created after the rescale use the new one.

**Problem**: The pre-write topology of the primary-key table Flink sink uses bucket shuffle,
and the numBuckets used for sharding is the **table-level value captured at job submission
time** (a constructor argument of `FlinkRowDataChannelComputer`). It does not follow the actual
per-partition bucket counts after a rescale, and there is no test coverage for this scenario.

**Failure chain**:
1. After a rescale, some partitions have an actual bucket count C ≠ the fixed value N. Records
with the same bucket key (same `hash % C`) are scattered across multiple subtasks (the
sharding key `hash % N` mismatches the actual bucket `hash % C`), so **one bucket is written
by multiple writers concurrently**.
2. The `UndoRecoveryOperator` instance upstream of each writer subtask maintains its own
`bucketOffsets` (the bucket's log end offset, reported after every successful
upsert/delete). The same bucket appears in multiple operator instances' maps with values
that are almost certainly different (the LEO grows monotonically and each writer's last
report point differs).
3. At checkpoint time, each operator instance emits one WriterState fragment — the same bucket
appears in multiple fragments with different offsets.
4. On job failure, `RecoveryOffsetManager.putMergedOffset` detects the conflicting per-bucket
offsets during restore and throws `IllegalStateException("Conflicting checkpoint offsets
...")`.

**Result**: after a partitioned table is rescaled, the affected Flink sink jobs can no longer
recover from checkpoint/savepoint on any failover — the only option is dropping the state and
starting over.

### Solution

Restore the "one bucket, exactly one writer" invariant: make the pre-write bucket shuffle
aware of the partition's **actual** bucket count.

`FlinkRowDataChannelComputer` no longer uses the table-level numBuckets captured at job
submission time. Instead it resolves the per-partition count at runtime and caches it locally
in the partitioner — a partition's bucket count is immutable once created, so a cached entry
is correct forever and needs no invalidation.

On a cache miss, the resolution follows three paths depending on the lookup result:
1. **Partition exists**: take the authoritative bucket count from the partition metadata and
cache it.
2. **Partition not yet created** (normal timing for dynamic partition creation — the creation
is triggered by the downstream writer, and blocking in the partitioner would deadlock):
optimistically group with the current table-level count — a new partition's bucket count is
exactly the table-level value at its creation moment; it is corrected once the partition
metadata is ready. The final bucket id is computed authoritatively by the writer from the
actual count, with server-side validation as the safety net, so no wrong data is produced
in this window.
3. **Metadata query failure**: after bounded retries, throw (fail fast) — no silent
degradation.

This only protects future checkpoints; WriterState that has already been fragmented cannot be
repaired — restoring from such checkpoints will still fail and they must be discarded.

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Start with FlinkRowDataChannelComputer and trace how its bucket shuffle resolves numBuckets, then inspect UndoRecoveryOperator and RecoveryOffsetManager to understand checkpoint state conflicts. Add coverage for a partitioned table rescaled to per-partition bucket counts, and verify that recovery no longer sees conflicting offsets while dynamic partition creation and metadata failures follow the stated behavior.

Written by the indexing model from the issue text.