[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 can write records whose original Paimon partitions have expired through the shared Fluss historical partition while preserving each record's original partition identity. This PR covers client routing, KV and log RPCs, historical KV recovery and lake lookup refresh, and tiering back to the original Paimon partitions.

Why resolve the write target before sending

Historical fallback changes the physical TableBucket from the original partition to the shared historical partition. Writer idempotence is scoped by physical TableBucket, so these targets have independent writer-ID and batch-sequence namespaces.

If the client always sends to the original partition first and waits for the server to report that it no longer exists, the batch has already entered the original target's sequence namespace. Other requests for the same target may also be in flight. Moving queued batches at that point requires pausing the old target, resolving its outstanding responses, preserving order, and assigning new writer IDs and sequences in the historical target. Until that handoff is implemented, an in-flight request makes transparent rerouting unsafe and the client must abort the affected target, which causes a Flink job failover.

To reduce the probability of reaching that case, the client resolves a potentially expired target before enqueueing the record. The client-side retention boundary is only a low-cost hint; refreshed metadata confirms whether the original partition still exists. If it exists, the write remains on the original target. If it is confirmed missing, the batch is created directly in the historical target's idempotence namespace.

This precheck narrows but cannot eliminate the race between metadata resolution and sending. If the server explicitly reports that the original target disappeared during that window, the client reroutes queued batches when no request for that target remains in flight. Otherwise, it aborts only the affected target to preserve idempotence and sequence correctness. The durable retirement and graceful in-flight handoff are tracked by #3820 and #4166; #4161 tracks replacing the client-side retention calculation with periodically refreshed, server-authoritative lifecycle metadata.

Brief change log

• Resolve potentially expired partitions before enqueueing, keep writes on an original partition that refreshed metadata still reports as present, and route confirmed-missing targets through the shared historical partition.
• Reroute queued batches after an explicit missing-partition response when the original target has no in-flight requests; otherwise abort only the affected target.
• Carry the original partition name through PUT_KV and PRODUCE_LOG, version-gate the new PRODUCE_LOG fields, and correlate responses by the actual physical target plus the original partition.
• Reject PRODUCE_LOG for primary-key tables so all primary-key writes, including historical writes, go through PUT_KV.
• Recover the local historical KV overlay from the lake-covered offset plus the remaining local WAL, and consult the lake lookuper on local misses.
• Refresh registered lake files in place when the required lake snapshot changes so unchanged downloaded lookup files remain reusable.
• Tier historical KV and append-only records back to their original Paimon partitions.
• Add unit and integration coverage for routing, RPC compatibility, missing-target handling, recovery, lookup refresh, and Paimon tiering.

Tests

Validation and targeted suites run during development:

• ./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
• RPC: ServerConnectionTest
• Client: ClientRpcMessageUtilsTest, write-batch tests, RecordAccumulatorTest, and SenderTest
• Server: HistoricalPartitionTableValidationTest, ServerRpcMessageUtilsTest, HistoricalPartitionTaskExecutorTest, and HistoricalPartitionManagerTest
• Paimon: PaimonTieringTest, PaimonLakeTableLookuperTest, and HistoricalPartitionITCase
• Regenerated the checked-in Rust protobuf bindings and ran the Rust workspace build, formatting check, and Clippy.

The complete ReplicaManagerTest suite could not finish locally because host disk usage exceeded the TabletServer test write limit and caused DiskWriteLockedException; CI provides the full validation.

API and Format

• Bumps PRODUCE_LOG from version 0 to version 1 and adds optional original_partition_name fields to bucket requests and responses. 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.
• The changed PaimonWriteResult serialization is used only for shuffle data, not persisted operator state, so no state-compatibility migration is required.
• No new public Java API or server configuration option is introduced.

Documentation

Updates the historical-partition table option description to cover write access.

Production hardening is tracked separately:

• Incremental historical RocksDB KV cleanup: [FIP-28] Support incremental cleanup of historical KV state #4159
• Durable partition freeze and retirement: [server] add lake-aware partition drop when auto-partition retention is enabled #3820
• Server-authoritative client partition metadata: [FIP-28] Refresh server partition status for historical write routing #4161
• Graceful handoff with in-flight requests: [FIP-28] Smoothly reroute writes with in-flight requests after partition retirement #4166

[Copilot]

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

[luoyuxia]

Thanks for all the review feedback. After addressing the comments, I believe the following follow-up work is still needed before historical partition writes are production-ready.

Remaining production-readiness work

1. P0 — Incremental TTL cleanup for the historical RocksDB KV overlay (#4159)

   Historical values and tombstones should carry their WAL offsets, while a monotonic cleanup watermark advances with confirmed lake progress. RocksDB compaction filters and rate-limited compaction can then reclaim lake-covered state incrementally, without waiting for the whole bucket to catch up or dropping and rebuilding the complete KV store.

2. P1 — Durable partition freeze and retirement protocol (#3820)

   Before auto-retention removes a hot partition, the server should durably freeze it, reject new writes, drain already accepted writes, record a final barrier for every bucket, and wait until a readable lake snapshot covers those barriers. Only a deterministic RETIRED state should permit historical routing and deletion of the original partition.

3. P1 — Periodic, server-authoritative partition metadata refresh (#4161)

   The client currently uses its retention calculation as a hint and refreshes metadata for a potentially expired partition. That calculation is not authoritative, may differ because of time-zone or stale-cache effects, and is too expensive on the per-record path. The client should instead periodically fetch batched, rate-limited partition lifecycle information from the server and use that as the routing source of truth.

4. P2 — Smooth handoff when the original target still has in-flight requests (#4166)

   Today, if the server explicitly reports that the original partition is missing while requests for that target are still in flight, the client aborts the affected target. For Flink, this causes a job failover. The race window should be small, but removing the failover requires coordinated idempotence handling: pause draining to the old target, resolve its outstanding requests, preserve ordering, and allocate fresh writer/sequence state in the historical physical bucket. The durable freeze/retirement protocol above should provide the deterministic transition needed to make this handoff tractable.

[luoyuxia]

@wuchong Thanks for you review. Comments have been addressed.

[wuchong]

Should we use a volatile boolean isHistoricalPartition? The original partition name is determinist when the batch is created which is in the physicalTablepath.

[wuchong]

Add an info log to record which partitions have transitioned to historical partitions.

[wuchong]

LGTM. I appended a commit to refactor WriteBatch a bit.

[luoyuxia]

@wuchong Thanks for the commit. LGTM! I rebase main branch to resolve conflict.