[server][client] Support per-partition bucket count for partitioned tables

[Kaixuan-Duan]

Purpose

Linked issue: close #3907

Support per-partition bucket count for partitioned tables. After this change, ALTER TABLE ... SET ('bucket.num' = N) only affects the default for newly created partitions; existing partitions retain their original bucket count (bucket.num.actual) in their own PartitionRegistration. This enables online bucket rescaling without disrupting data already written to existing partitions.

The PR also introduces bucketLayoutEpoch, a table-level monotonically increasing version for bucket-layout changes, to detect stale client metadata and return STALE_METADATA when a request's bucket count does not match the server's current state.

Lake table is currently only implemented for Paimon.

Brief change log

• Migrate bucket-count ownership from PhysicalTablePath to TablePartition across client, server, and connector modules so that bucket counts and BucketAssigners are keyed by the immutable TablePartition(tableId, partitionId).
• Add bucketLayoutEpoch to TableRegistration, propagated through PbTableMetadata and GetTableInfoResponse; withBucketCount atomically replaces the table-level count and increments the epoch.
• Add a per-table read/write lock so ALTER bucket.num and partition creation cannot interleave; partition creation always reads the fresh bucket count from ZK.
• Change client-side dynamic partition creation from async to synchronous — bucket assignment blocks until the partition's partitionId and actual bucket count are present in the client metadata.
• Commit the new table-level count, bucketLayoutEpoch + 1, and legacy-partition bucketCount backfill in one version-checked, epoch-fenced ZK transaction.
• Add bucket_count to all bucket-routed request protos and validate it server-side; mismatch returns STALE_METADATA, which triggers metadata + BucketAssigner invalidation and batch failure on the client.
• Add epoch-aware fallback in PartitionRegistration.getBucketCountOrDefault: use the persisted count when present, fall back to table-level when epoch is 0, throw StaleMetadataException when epoch > 0 and the count is missing.
• Read table metadata before partitions in listPartitionInfos to prevent cross-ALTER inconsistency.
• Reject historical partition lookup on rescaled tables (bucketLayoutEpoch > 0) with a TODO for future support.
• Propagate bucket.num changes to the Paimon lake catalog separately from schema-alter branches.

Tests

./mvnw clean verify -pl fluss-common,fluss-rpc
./mvnw clean verify -pl fluss-client
./mvnw clean verify -pl fluss-server
./mvnw clean verify -pl fluss-flink/fluss-flink-common,fluss-flink/fluss-flink-tiering
./mvnw clean verify -pl fluss-lake/fluss-lake-paimon
./mvnw clean verify -pl fluss-lake/fluss-lake-iceberg,fluss-lake/fluss-lake-hudi,fluss-lake/fluss-lake-lance
./mvnw clean verify -pl fluss-spark/fluss-spark-common,fluss-spark/fluss-spark-3.5
./mvnw clean verify -pl fluss-spark/fluss-spark-ut
cargo fmt --all -- --check
cargo check -p fluss-rs

API and Format

• Proto: Added optional int64 bucket_layout_epoch to PbTableMetadata and GetTableInfoResponse; optional int32 bucket_count to all bucket-routed request messages. All fields are optional for backward compatibility with old servers/clients.
• ZK JSON serialization includes the new bucketCount / bucketLayoutEpoch fields; old data without them is deserialized with null/0 defaults.

Documentation

• Updated DDL docs (ddl.md) — ALTER TABLE bucket.num semantics
• Updated options docs (options.md) — bucket.num option description
• Updated bucketing docs (bucketing.md) — per-partition bucket count behavior

[Copilot]

Pull request overview

This PR implements per-partition bucket counts for partitioned tables by moving “actual bucket count” ownership to partitions, introducing a table-level bucketLayoutEpoch to detect stale client metadata, and propagating bucket-count validation across server/client/connectors/lake integrations. It updates RPC/metadata serialization paths and tightens client behavior so bucket assignment cannot proceed until the partition’s bucket metadata is known.

Changes:

• Add bucketLayoutEpoch to table metadata and use it to detect/resist stale bucket layouts (STALE_METADATA).
• Persist and propagate per-partition bucket counts (bucket.num.actual) and update Flink/Spark/lake tiering logic to enumerate buckets per-partition.
• Add bucket_count to bucket-routed RPCs and validate it on the TabletServer to fail fast on stale client routing.

Reviewed changes

Copilot reviewed 110 out of 110 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
website/docs/table-design/data-distribution/bucketing.md	Document ALTER bucket.num semantics for partitioned tables
website/docs/engine-flink/options.md	Clarify bucket.num option semantics (new partitions only)
website/docs/engine-flink/ddl.md	Document ALTER TABLE ... SET ('bucket.num' = ...) behavior/limits
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/read/SplitPlannerLakeBucketGuardTest.scala	Spark guard test for out-of-range lake buckets
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeTableReadTestBase.scala	Spark lake read sync uses per-partition bucket counts
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala	Spark micro-batch offset enumeration per-partition
fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java	Update ZK partition registration tests with bucketCount arg
fluss-server/src/test/java/org/apache/fluss/server/zk/data/TableRegistrationJsonSerdeTest.java	Include bucket_layout_epoch in expected table JSON
fluss-server/src/test/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerdeTest.java	Partition registration v2 JSON + bucket_count backward-compat test
fluss-server/src/test/java/org/apache/fluss/server/testutils/PartitionMetadataAssert.java	Avoid failing legacy assertions when expected bucketCount is null
fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java	Enumerate snapshot buckets using per-partition bucket counts
fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java	Update partition registration helper calls with bucket count
fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java	Update partition metadata registration calls with bucket count
fluss-server/src/test/java/org/apache/fluss/server/coordinator/event/watcher/TableChangeWatcherTest.java	Add ZK version usage in watcher test + updated registration calls
fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java	Add ZK version parameter in table-change tests + registration updates
fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java	Add versioned reads + atomic CAS+epoch-fenced backfill transaction
fluss-server/src/main/java/org/apache/fluss/server/zk/data/TableRegistrationJsonSerde.java	Serialize/deserialize bucket_layout_epoch
fluss-server/src/main/java/org/apache/fluss/server/zk/data/TableRegistration.java	Add bucketLayoutEpoch and atomic withBucketCount()
fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistrationJsonSerde.java	Add optional bucket_count and bump serde version to v2
fluss-server/src/main/java/org/apache/fluss/server/zk/data/PartitionRegistration.java	Persist per-partition bucket count + epoch-aware fallback/guard
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java	Propagate bucketLayoutEpoch + partition bucketCount in RPC metadata
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java	Validate request bucket_count and throw STALE_METADATA on mismatch
fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java	ListPartitionInfos reads table first + use per-partition bucketCount
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java	Reject historical lookup on rescaled tables (epoch > 0)
fluss-server/src/main/java/org/apache/fluss/server/metadata/ZkBasedMetadataProvider.java	Populate PartitionMetadata with bucket count
fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerMetadataSnapshot.java	Track partition bucket counts + epochs; validate request bucket_count
fluss-server/src/main/java/org/apache/fluss/server/metadata/PartitionMetadata.java	Add optional bucketCount field
fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java	Populate PartitionMetadata with effective bucket count
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java	Include per-partition bucketCount in UpdateMetadata
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java	Expose coordinator ZK version + refresh auto-partition tables on bucket change
fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java	Guard partition creation with rescale locks + read fresh bucket count from ZK
fluss-rust/crates/fluss/src/rpc/message/put_kv.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/produce_log.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/prefix_lookup.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/lookup.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/list_offsets.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/rpc/message/limit_scan.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/metadata/table_stats.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/metadata/partition.rs	Add bucket_count field (legacy None for now)
fluss-rust/crates/fluss/src/client/table/scanner.rs	Add bucket_count to fetch-log requests (legacy None for now)
fluss-rpc/src/main/proto/FlussApi.proto	Add bucket_layout_epoch + bucket_count optional fields for compatibility
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java	Add STALE_METADATA error mapping to StaleMetadataException
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java	Verify partition bucket-count stamping across tiering rounds
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/PaimonLakeCatalogTest.java	Ensure user-facing Paimon bucket option remains rejected; Fluss bucket.num applies
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeWriter.java	Override Paimon BUCKET for fixed-bucket tables using resolved actual bucket count
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java	Refactor writer creation helper
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeCatalog.java	Apply Fluss bucket.num changes directly to Paimon BUCKET option
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java	Provide WriterInitContext.bucketCount implementation in tests
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java	Provide WriterInitContext.bucketCount implementation in tests
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java	Reject bucket.num rescale for Iceberg (unsupported)
fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringTest.java	Provide WriterInitContext.bucketCount implementation in tests
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkTestBase.java	ZK partition registration includes bucket count
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContextTest.java	Validate partition bucket count handling (fallback vs fail-loud)
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java	PartitionInfo now includes bucketCount in test fixtures
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManagerTest.java	Recovery enumerates buckets per-partition; adds rescale-focused tests
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/lake/LakeSplitGeneratorTest.java	Flink guard test for out-of-range lake buckets
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java	Update expectations: server rejects non-partitioned bucket.num alter
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java	Count(*) enumerates buckets per-partition
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java	Add WriterInitContext.bucketCount resolution/fail-loud for partitioned
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java	Snapshot per-partition bucket counts for correct lake writer stamping
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitGenerator.java	Generate tiering splits per-partition bucket count
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlussOnlyBatchSplitGenerator.java	Generate log splits per-partition bucket count
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java	Carry partition bucketCount through enumerator partition model
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManager.java	Enumerate offsets and buckets per-partition bucket count
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java	Use per-partition bucket counts; add fail-loud union-read guard
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java	Allow ALTER bucket.num; update option description
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/OrphanCleanUtils.java	Enumerate buckets using PartitionInfo bucketCount fallback
fluss-common/src/test/java/org/apache/fluss/lake/writer/WriterInitContextTest.java	Implement new WriterInitContext.bucketCount in tests
fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java	Add bucketLayoutEpoch to core TableInfo metadata
fluss-common/src/main/java/org/apache/fluss/metadata/PartitionInfo.java	Make bucketCount a first-class field; add helper bucketCountOrDefault
fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java	Add bucketCount() to writer init contract
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeCatalog.java	Document bucket.num SetOption contract for lake catalogs
fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java	Track bucket counts by TablePartition/tableId; invalidate with metadata
fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java	Test STALE_METADATA handling and BucketAssigner invalidation
fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java	Test parsing partition bucket_count with fallback
fluss-client/src/test/java/org/apache/fluss/client/table/PartitionedTableITCase.java	Update expectations: dynamic partition creation is synchronous now
fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupSenderTest.java	Ensure lookup requests carry pinned bucketCount and legacy omit behavior
fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java	End-to-end test: Kv snapshots reflect per-partition bucket counts
fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java	Key BucketAssigner cache by TablePartition/tableId; pass bucketCount into batches
fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java	Store bucketCount used to compute bucketId for request validation
fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java	Fail batch on STALE_METADATA; invalidate metadata + BucketAssigner
fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java	Carry bucketCount into newly created write batches
fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java	Make dynamic partition creation synchronous and metadata-aware
fluss-client/src/main/java/org/apache/fluss/client/utils/MetadataUtils.java	Rebuild Cluster with per-partition/table bucketCount maps
fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java	Add bucket_count to routed requests; parse PartitionInfo bucket_count with fallback
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java	Enumerate batch scan buckets per partition bucketCount
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java	Include bucket_count in fetchLog requests
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LimitBatchScanner.java	Include bucket_count in limit scan requests
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java	Include bucket_count in scan requests
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java	Re-route bucketId by per-partition bucketCount; reject historical lookup on rescaled tables
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixLookupQuery.java	Carry bucketCount through prefix lookup query
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixLookupBatch.java	Store pinned bucketCount for prefix batch requests
fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixKeyLookuper.java	Compute bucketId after resolving per-partition bucketCount
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java	Fail fast on STALE_METADATA; keep invalid-metadata refresh behavior
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupQuery.java	Carry bucketCount through lookup query
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupClient.java	Plumb bucketCount into lookup/prefixLookup
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupBatch.java	Store pinned bucketCount for batch requests
fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookupQuery.java	Add bucketCount field to lookup query base
fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookuper.java	Add helper to resolve per-partition bucketCount from cluster metadata
fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java	Parse bucketLayoutEpoch; resolve partition bucketCount fallback with epoch guard; include bucket_count in stats/offset requests

Suppressed comments (2)

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java:1365

• This comment says "ALTER bucket.num 4 -> 8" but the test variables change 2 -> 4 (originalBucketNum=2, newBucketNum=4). Keeping these numbers accurate is important for understanding what the test asserts.

        // ALTER bucket.num 4 -> 8.

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java:1374

• This comment says the new partition uses bucket.num.actual = 8, but newBucketNum is 4. The comment should reflect the actual post-ALTER bucket count used by the test.

        // New partition created AFTER the ALTER uses bucket.num.actual = 8.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Kaixuan-Duan]

@luoyuxia Thanks for the review. I've addressed your comments. Could you please take another look?

[Kaixuan-Duan]

@platinumhamburg Thanks for the review. I've addressed comments. Could you please take a look?

[platinumhamburg]

HistoricalPartitionLookupITCase consistently gets stuck after the original partitions are deleted: the newly added bucket-count validation returns TABLET_METADATA_NOT_READY before the existing lookup path can report PartitionNotExistException and trigger the historical-partition fallback. The client then retries indefinitely until the CI job times out.

[Kaixuan-Duan]

@platinumhamburg That's right. I've fixed this infinite retries bug by choosing to trigger a historical partition rollback. Let's see the CI results.

[platinumhamburg]

This striped-lock design came from my earlier review feedback and was refined over several rounds. After reviewing the complete implementation again, I think I should revise that recommendation: we should remove bucketRescaleLocks entirely, rely on ZooKeeper OCC, and reduce the complexity and entropy introduced by this PR.

Lock acquisition is now spread across explicit partition creation/deletion, auto-partitioning, and historical-partition paths, while still not covering dropTable, createTable, or cross-JVM concurrency after coordinator failover. It therefore cannot be the actual correctness boundary, but every related DDL path must still follow the same locking and lock-ordering rules. The fair ALTER write lock also spans lake propagation, partition enumeration, and multiple ZooKeeper reads, blocking both same-table operations and unrelated tables that hash to the same stripe.

The core invariant is already protected by ZooKeeper: new partitions atomically persist their assignment and explicit bucketCountActual, while the legacy-partition backfill and table-level bucket/epoch update already use per-partition version CAS, table-version CAS, and the coordinator-epoch fence. A partition creation overlapping an ALTER still persists the actual count used to build its assignment, so it does not require an additional read lock.

I suggest removing all bucketRescaleLocks acquisition sites. On BadVersion or a concurrent-delete NoNode, ALTER should re-read the metadata and retry with a bound; if the table is gone, return a clear TableNotExistException. Lake propagation should participate in the same OCC retry: after each CAS conflict, re-run the idempotent propagation from the latest state before committing to ZooKeeper. This makes CAS/transactions the single concurrency mechanism, without adding sequence counters, generation fencing, an outbox, or a persistent DDL state machine, and materially reduces both runtime contention and implementation complexity.

Also, this PR needs to be rebased.

[Kaixuan-Duan]

@platinumhamburg Thanks for the thorough re-review.
Removed bucketRescaleLocks entirely (all 9 acquisition sites, the stripe array, and the lock-ordering rules). Concurrency now relies solely on ZooKeeper OCC: BadVersion/NoNode retried with a bound from the latest state, table-gone returns TableNotExistException, and the idempotent lake propagation re-runs inside the same retry.

[platinumhamburg]

Blocking: routing metadata should be part of replica activation instead of requiring clients to compensate for split readiness

The current design separates replica readiness from routing metadata readiness.

A TabletServer may process NotifyLeaderAndIsr, create the replica, and become the leader before a separate UpdateMetadata request populates bucketCountActual in ServerMetadataSnapshot. This PR then exposes that internal timing window as TABLET_METADATA_NOT_READY and adds client-side retry logic to wait for the second metadata path to catch up.

However, once bucketCountActual is required to validate client routing correctly, it is no longer optional auxiliary metadata. It should be part of the replica activation state.

The Coordinator already has the complete assignment for the relevant table or partition when constructing NotifyLeaderAndIsr. I suggest:

• carrying bucketCountActual in NotifyLeaderAndIsr;
• storing it as part of the replica’s routing state;
• completing leader activation only after the replica has this state;
• resolving the leader replica before validating the request’s routingBucketCount against the replica-local actual count.

With this model, request outcomes become straightforward:

• the table, partition, or replica does not exist: preserve the existing not-exist or unknown-object error;
• the replica exists but is not the leader: return NOT_LEADER_OR_FOLLOWER;
• the leader replica exists but the routing count differs: return STALE_METADATA;
• the leader replica exists and the count matches: execute the request.

UpdateMetadata can still support metadata discovery and client metadata refresh, but its arrival should not determine whether an already activated leader can process requests.

This would remove or substantially simplify several pieces of complexity introduced by the current approach:

• the normal-path semantics of TabletMetadataNotReadyException;
• the scheduler and custom retry loop added to FlussAdmin;
• hard-coded retry deadlines and shutdown/in-flight future edge cases;
• ambiguity between incomplete metadata propagation, missing leadership, and deleted objects;
• masking of TableNotExistException and PartitionNotExistException;
• duplicated routing-readiness state in ServerMetadataSnapshot.

I suggest establishing the following invariant:

A ready leader must also have its routing bucket count ready.

Client operations should not need to retry while two independent TabletServer metadata paths eventually converge.

[platinumhamburg]

I do have several concerns about the current test suite:

1. ReplicaManagerTest directly calls Replica.updateRoutingState to simulate the post-ALTER state. ALTER bucket.num does not perform this transition in production; it advances the TabletServer metadata cache through UpdateMetadata. This test should activate the leader at epoch 0, advance the cache to epoch 1 through ReplicaManager.maybeUpdateMetadataCache, and then verify that a request without routingBucketCount is rejected. Otherwise the test masks the stale-epoch problem instead of reproducing it.

2. testAlterAndConcurrentPartitionCreationConvergeUnderOcc does not deterministically establish the concurrency described by its name and comments. The latch only orders thread startup, so the test may pass with two sequential operations without exercising an OCC conflict. Please either make the interleaving deterministic through an existing test seam and verify that the partition assignment size matches its persisted bucketCountActual, or remove this test if that production race cannot be expressed reliably. Please do not add a production-only hook solely for this test.

[platinumhamburg]

Could we revert the asynchronous dynamic-partition write changes from this PR and keep the original blocking slow path for now?

The per-partition bucket-count change requires us to wait until the partition ID and bucketCountActual are available and then route/append using a consistent metadata snapshot. It does not inherently require deferred writes.

The current asynchronous path introduces a second write lifecycle outside RecordAccumulator: copied records are held in unbounded future chains, flush() and close() must coordinate those chains and their executor, and the continuation waits for metadata but then re-enters doSend() and reads mutable metadata again. In particular, the partition-metadata executor is not shut down by close(), and invalidation between the readiness check and the later metadata read can still make an accepted write fail.

A blocking slow path is considerably easier to make correct here: use the local metadata fast path for existing partitions; otherwise create/check the partition and synchronously wait for a routing snapshot containing the partition ID and actual bucket count; then perform bucket assignment and append with that same snapshot. The existing blocking behavior also provides conservative backpressure and preserves the current ownership, ordering, flush, and close semantics.

If non-blocking progress across partitions is shown to be important, I think it should be handled as a separate follow-up with explicit requirements and tests for bounded memory, per-partition ordering, timeout, flush, close, and worker termination. That keeps this PR focused on per-partition bucket-count correctness.

[Kaixuan-Duan]

@platinumhamburg Thanks for the suggestion. I’ve reverted the deferred-write path and implemented the blocking slow path as proposed. Existing partitions use the local metadata fast path; otherwise, the writer waits until both the partition ID and bucketCountActual are available, then uses the same metadata snapshot for routing and appending.

[wuchong]

I just merge another big PR and introduces some conflicts with this PR. Please also rebase and resolve the conflicts. Thanks.

[platinumhamburg]

Hi @Kaixuan-Duan, I pushed one commit to this PR branch.

The problem. The routing bucket count was validated per request: a single
bucket carrying a stale count failed the entire request. But a rescale only
changes newly created partitions, so the co-batched buckets of the other
partitions were still routed correctly and got taken down with it. Since
FetchLogRequest spans several tables, one stale bucket failed the reads of all
of them; on the write side produceLog/putKv lost the whole table's batch.

The approach. Report a stale route per bucket rather than per request — the
same way authorizeRequestData already reports authorization failures. A single
shared helper holds the decision, and it stays allocation-free when every bucket
is correctly routed, so the common request path is unchanged. limitScan and
scanKv carry one bucket and listOffsets carries a request-scoped count, so
failing those as a whole is already exact — they are left untouched.

One request on the git side: now that this PR is under review, please avoid
squashing the branch.

[Kaixuan-Duan]

@platinumhamburg Thanks for the commit. Reporting stale routes per bucket is the right fix — failing the whole multi-table fetchLog request over one stale bucket was too broad, and reusing the authorizeRequestData pattern keeps the error channel consistent.
I'll keep the branch history intact and won't squash while the PR is under review.