[client][server][paimon] Complete historical partition writes

[luoyuxia]

Generated-by: Codex (GPT-5) following the guidelines

Purpose

Linked issue: close #4119

Complete the end-to-end historical partition write support left as follow-up work by #4001. Writers should accept records for expired Paimon partitions, preserve each record's original partition identity while using the shared Fluss historical partition, and safely reclaim fully tiered local KV overlays.

Why resolve the write target before sending

A historical fallback changes the physical TableBucket from the original partition to the shared historical partition. Fluss writer idempotence is scoped by physical TableBucket, so the original and historical targets use different writer ID and batch-sequence namespaces. The original sequence cannot simply be reused, especially because multiple original partitions may share the same historical bucket.

Once a request has been sent to the original target, transparently switching it to the historical target would require tracking all related in-flight requests, pausing new sends, preserving queue order, distinguishing deterministic failures from ambiguous outcomes, and reassigning sequences in the new physical bucket. This would add substantial state-management and concurrency complexity to the client.

This PR therefore uses a simpler approach. The client uses the auto-partition retention boundary only as a low-cost filter for partitions that may have expired, then refreshes metadata to confirm that the original partition no longer exists before enqueueing the record. This avoids metadata requests for current and future partitions and prevents the dynamic partition creator from recreating an expired partition. In the narrow race where the partition exists during this check but is removed before the server handles the request, the request fails through normal response handling. For a Flink job, this triggers failover; after the job restarts, the new writer sees that the original partition is missing and routes the replayed records to the historical partition.

Brief change log

• Resolve potentially expired partitions before enqueueing write batches, route confirmed historical writes through the internal historical partition, and fail batches when refreshed metadata confirms that the original target no longer exists.
• Carry the original partition name through PUT_KV and PRODUCE_LOG, version-gate the new PRODUCE_LOG fields, and correlate responses by physical bucket plus original partition.
• Tier historical KV and log records back to their original Paimon partitions, including the recovery path.
• Throttle oversized historical KV overlays and clean fully tiered overlays after the configured idle period, with leader-epoch and log-end-offset guards around asynchronous cleanup.
• Add unit and integration coverage for routing, RPC compatibility, tiering, recovery, missing targets, dynamic configuration, and cleanup.

Tests

• ./mvnw -nsu -DskipTests -DskipITs -pl fluss-common,fluss-rpc,fluss-server validate
• ./mvnw -nsu -DskipTests -DskipITs -pl fluss-client,fluss-flink/fluss-flink-common,fluss-lake/fluss-lake-paimon validate
  • Checkstyle and Spotless passed for all affected modules.
• Targeted RPC test: ServerConnectionTest (4 tests passed).
• Targeted server tests: DynamicConfigChangeTest, HistoricalPartitionTableValidationTest, ServerRpcMessageUtilsTest, HistoricalPartitionTaskExecutorTest, and HistoricalPartitionManagerTest (55 tests passed).
• Targeted client tests: ClientRpcMessageUtilsTest, ArrowLogWriteBatchTest, CompactedLogWriteBatchTest, IndexedLogWriteBatchTest, KvWriteBatchTest, RecordAccumulatorTest, and SenderTest (78 tests passed).
• Targeted Paimon test: PaimonTieringTest (22 tests passed).
• Targeted Paimon integration test: HistoricalPartitionITCase (4 tests passed, including leader restart and lookup fallback to Paimon).
• Regenerated the checked-in Rust protobuf bindings with fluss-rust/crates/fluss/regen.sh; generation is idempotent, cargo build --workspace --all-targets --exclude fluss_python --exclude fluss-cpp --exclude fluss_nif, cargo fmt --all -- --check, and workspace clippy with -D warnings passed.
• ReplicaManagerTest could not complete locally because the host disk usage was about 91.6%, above the TabletServer test write limit of 85%; the resulting DiskWriteLockedExceptions caused cascading failures. This suite is left to CI.

API and Format

• Bumps PRODUCE_LOG from version 0 to version 1 and adds optional original_partition_name fields to bucket requests (field 4) and responses (field 6). Version gates keep version 0 peers compatible.
• Reuses the optional original-partition field introduced for PUT_KV by [server] Support primary-key writes to historical partitions #4001.
• Adds the dynamic server option server.historical-partition.kv-cleanup.idle-time, defaulting to 30 minutes; setting it to 0 disables idle cleanup.
• No new public Java API is introduced.

Documentation

Updates the historical-partition table option description to cover write access. No standalone user documentation is added in this PR.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.