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

Search before asking

• I searched in the 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?

• I'm willing to submit a PR!