[client] Resolve dynamic partition bucket counts before sending

[wuchong]

Purpose

Refs #4287.

A writer may buffer records before a dynamically created partition's bucket count is known. Resolve the partition ID and bucket count together before sending, so a stale table default cannot silently route records to the wrong buckets.

This PR covers tentative routing resolution and server-reported routing failures. The existing synchronous partition-existence check and shared asynchronous metadata refresh work remain follow-ups.

Brief change log

• Let RecordAccumulator own each physical partition's assigner, temporary bucket count, and resolved layout. Pass the current count to assignBucket and onNewBatch.
• When the actual count differs, retain buffered batches only for non-hash assigners with every occupied bucket ID below the new count. Otherwise, fail all tentative batches for that context with BucketRescaleException; callers can resubmit the failed records with the same writer.
• Keep memory allocation and failure callbacks outside routing locks. Synchronize tentative routing resolution and historical target switches while preserving independent writes to resolved buckets.
• Let the server validate resolved batches using the count carried in each request; remove the redundant metadata comparison and invalidBucketRoutingTables result from the send path. Treat server-reported INVALID_BUCKET_ROUTING and partition-creation failures as fatal writer errors, including queued/in-flight batch cleanup and propagation of the original failure to later writes.
• Simplify historical routing using the existing restriction that historical partitions and bucket rescaling cannot be enabled together.

Tests

• Passed 154 focused client tests across BucketRoutingTest, RecordAccumulatorTest, SenderTest, HashBucketAssignerTest, StickyStaticBucketAssignerTest, PartitionBucketCountRescaleITCase, PartitionedTableITCase, and AutoPartitionedTableITCase, including sending the original resolved count after metadata changes and handling server rejection for both count growth and shrinkage.
• Coverage includes growth/shrinkage, no-key retention and out-of-range rejection, retries with the same writer, consistent metadata snapshots, buffer exhaustion, independent bucket appends, historical routing, and close/fatal-error races.
• Passed ./mvnw -B spotless:check validate.
• Full ./mvnw -B verify stopped in fluss-server: ReplicaTest.testBucketPhysicalStorageLocalLogSizeIncludesFollower and testPhysicalStorageLocalLogSizeIsScopedPerBucket both reported expected 450 but actual 7220. No server files are changed by this PR. An isolated rerun of all 16 ReplicaTest cases passed; the full-suite failure remains unresolved, and downstream modules were not verified by that run.

API and Format

Adds the public evolving BucketRescaleException for buffered records rejected before sending. Its recovery contract permits resubmission with the same writer. Bucket assigner signature changes are internal; RPC and storage formats are unchanged.

Documentation

The retry contract is documented in BucketRescaleException Javadoc. No standalone documentation changes.

Generative AI assistance: OpenAI Codex (GPT-6 Astra).