[server] feat: remote manifest v2

[gyang94]

Purpose

Linked issue: close #3835

Brief change log

1. Manifest V2 with derived logical view
2. Replacement planner
3. Atomic publish via CAS
4. Readers honor logical bounds; old clients keep working

Tests

API and Format

Documentation

[Copilot]

Pull request overview

Adds “remote manifest v2” support across Fluss (server, RPC, clients) by introducing an authoritative, versioned manifest handle with CAS publishing, plus a FetchLog v1 response format that can carry logical remote-log segment references.

Changes:

• Introduce Manifest V2 handle semantics (generation + logical range hints) and ZooKeeper CAS APIs for publishing the authoritative handle.
• Add FetchLog API v1 response fields for logical remote segment references, and propagate request API version into server-side fetch decisions.
• Add Manifest V2 writer gating via tablet-server advertised capabilities, plus orphan/GC plumbing and related metrics; update Java/Rust clients to consume logical references.

Reviewed changes

Copilot reviewed 69 out of 69 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java	Adds CAS behavior tests for authoritative remote manifest handle in ZooKeeper.
fluss-server/src/test/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerdeTest.java	Updates serde tests for registration version bump + capabilities field; switches to AssertJ.
fluss-server/src/test/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandleJsonSerdeTest.java	Extends serde tests for handle v2 fields and strict version dispatch.
fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java	Adds RPC round-trip tests for manifest CAS commit request and FetchLog v1 remote metadata.
fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerTest.java	Asserts tablet server advertises the new remote-manifest capability.
fluss-server/src/test/java/org/apache/fluss/server/log/remote/TestingRemoteLogStorage.java	Adds knobs to simulate failures for new deletion/GC paths.
fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java	Adds tests for replacing v2 manifests and atomically updating manifest+handle.
fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java	Adds validation tests for manifest-handle hint consistency and FetchLog behavior after V2 migration.
fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java	Simulates CAS conflicts and routes CAS commits through ZK CAS APIs in tests.
fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java	Adds integration coverage for Manifest V2 commit results (committed/conflict/fenced) and writer gating.
fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java	Adds APIs for reading versioned handles and CAS/create-if-absent publishing.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/VersionedRemoteLogManifestHandle.java	Introduces (handle, zkVersion) wrapper for authoritative snapshot reads.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerde.java	Bumps registration JSON version and serializes/deserializes capabilities.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistration.java	Adds capability set + helper methods for registration records.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandleJsonSerde.java	Adds versioned serialization format for handle v1/v2 with strict field validation.
fluss-server/src/main/java/org/apache/fluss/server/zk/data/RemoteLogManifestHandle.java	Adds handle versioning + v2 hints (generation/start) with invariants.
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java	Encodes/decodes FetchLog v1 remote references and CAS commit request fields.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java	Passes per-request API version into FetchParams for response-format decisions.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java	Registers tablet server capability for manifest-version dispatch.
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Returns FetchLog v1 remote references when negotiated; rejects v0 reads against persisted V2 manifests.
fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java	Supports both remote fetch metadata versions and advances fetch offset using logical bounds.
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java	Adds counters/gauges for GC/orphan tracking.
fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerInfo.java	Adds capability set + supportsCapability to server metadata.
fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java	Reworks indexing to logical references; tracks unreferenced/orphan metrics; supports atomic manifest+handle replace.
fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogStorageObject.java	Introduces discovered orphan object descriptor (path + modification time).
fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogStorage.java	Extends storage SPI with list/delete operations used by orphan sweeper/GC.
fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManifestUpdatePlan.java	Adds immutable “plan” object tying computed V2 manifest to CAS base handle and commit request data.
fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManifestCommitter.java	Adds CAS commit + reconciliation logic and atomic local apply after authoritative publish.
fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java	Loads versioned authoritative handle, validates handle/manifest consistency, and exposes logical reference APIs.
fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java	Implements listing and deletion of orphan objects/snapshots.
fluss-server/src/main/java/org/apache/fluss/server/log/FetchParamsBuilder.java	Adds API version plumbing into FetchParams.
fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java	Stores negotiated API version for fetch path decisions.
fluss-server/src/main/java/org/apache/fluss/server/kv/RemoteLogFetcher.java	Switches remote scanning to use logical references and enforces logical-end boundaries.
fluss-server/src/main/java/org/apache/fluss/server/entity/RemoteLogManifestExpectedHandleState.java	Adds enum for CAS precondition state (ABSENT/PRESENT).
fluss-server/src/main/java/org/apache/fluss/server/entity/RemoteLogManifestCommitResult.java	Adds enum for commit outcomes (committed/conflict/fenced/invalid).
fluss-server/src/main/java/org/apache/fluss/server/entity/CommitRemoteLogManifestData.java	Adds CAS fields and builders for v2 commit requests with validation.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TabletServerChangeWatcher.java	Propagates capabilities into ServerInfo updates on tablet-server ZK changes.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java	Returns INVALID_MANIFEST when CAS request parsing fails instead of throwing.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java	Gates V2 commits on config + replica capabilities; implements CAS publish with explicit commit_result codes.
fluss-rust/crates/fluss/src/rpc/api_key.rs	Extends FetchLog supported version range to v1; adds regression test.
fluss-rust/crates/fluss/src/client/table/scanner.rs	Decodes v0 vs v1 remote metadata exclusively; threads logical bounds through pending remote fetches.
fluss-rust/crates/fluss/src/client/table/remote_log.rs	Adds v1 decoding producing logical ranges; validates contiguity and bounds.
fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs	Enforces logical-end bound while consuming remote batches; threads logical-end through pending/completed fetches.
fluss-rpc/src/main/proto/FlussApi.proto	Adds CAS fields to commit request/response and adds FetchLog v1 remote reference messages.
fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java	Decodes FetchLog v1 remote references and rejects ambiguous v0+v1 responses.
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java	Bumps FetchLog API max version to 1 and documents semantics.
fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/FetchLogResultForBucket.java	Adds RemoteLogFetchInfoV2 variant and updates fetchFromRemote() semantics.
fluss-common/src/test/java/org/apache/fluss/remote/RemoteLogSegmentTest.java	Adds unit tests for half-open (exclusive end) offset invariants.
fluss-common/src/test/java/org/apache/fluss/remote/RemoteLogManifestJsonSerdeTest.java	Extends manifest serde tests for V2 generation/start/unreferenced segments and stricter validation.
fluss-common/src/main/java/org/apache/fluss/remote/UnreferencedRemoteLogSegment.java	Adds persisted metadata for segments removed from logical reads (GC input).
fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegmentReference.java	Adds logical reference wrapper for physical segment + logical bounds.
fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java	Clarifies half-open offsets; enforces exclusive end > start.
fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestV2Migration.java	Adds deterministic migration logic from V1 manifest to canonical V2 form.
fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestReplacementPlanner.java	Adds overlap-aware planning for segment replacement/expiration into V2 manifests.
fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestJsonSerde.java	Adds V2 serde fields (generation/start/unreferenced) with strict validation and v1/v2 dispatch.
fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogFetchInfoV2.java	Adds v1 remote fetch metadata holding logical references.
fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java	Adds metric names for unreferenced/orphan/GC failure tracking.
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds config flags for V2 writer, GC enablement, and GC grace period.
fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java	Updates tests for exclusive end offsets.
fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteCompletedFetchTest.java	Adds tests for logical clipping and rejects logical-end inside a batch boundary.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemotePendingFetch.java	Threads logical end offsets into completed fetch construction.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteCompletedFetch.java	Adds constructor variant passing logical end to base CompletedFetch.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java	Adds handling for FetchLog v1 logical references and creates pending fetches with logical bounds.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java	Enforces logical-end bound during batch iteration for remote fetches.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[swuferhong]

I left some comments:

[swuferhong]

Thanks @gyang94 for such a big feature. I left some comments.

[swuferhong]

INVALID_MANIFEST needs to fall back to the V1 writer, otherwise we get an upload → reject → delete churn loop during upgrades.

Right now only CONFLICT gets special handling; INVALID_MANIFEST and FENCED both fall through to cleanupRejectedPlan, which deletes the segments we just uploaded:

But INVALID_MANIFEST is exactly what the coordinator returns when its own writer gate is still closed. So:

Scenario: a future release flips the default to true, and an operator upgrades the tablet servers first while the coordinator is still on the old version (gate closed).
→ TS sends a V2 commit → coordinator replies INVALID_MANIFEST → cleanupRejectedPlan deletes the freshly uploaded segments → next tiering interval repeats the whole thing. For the entire upgrade window we burn remote bandwidth plus PUT/DELETE request cost and tiering makes zero progress.

Suggestion: treat INVALID_MANIFEST as "the coordinator is not ready for V2 yet" rather than as a hard rejection — for this round, fall back to runOnceV1 for the bucket, and set a process-local flag so the next round probes V2 before committing to it. That makes the config-gated rollout genuinely order-independent: neither upgrade order causes churn, and we don't have to document a required upgrade sequence.

Please also add test coverage for the INVALID_MANIFEST and FENCED paths — neither is exercised today.

[swuferhong]

baseManifest and baseHandle are read through two separate lock acquisitions. If an old task overlaps with a newly started task, remoteLog may be updated between these reads, producing an inconsistent pair—for example, an old manifest with a newer ZK handle/version. The subsequent CAS could then succeed using the newer ZK version while publishing a manifest derived from stale state, potentially overwriting a more recent update.

Please expose an atomic snapshot from RemoteLogTablet that contains both the manifest and its versioned handle (like ManifestSnapshot), captured under the same read lock, and use that snapshot as the base for planning and CAS.

[swuferhong]

Interruption may not stop the task from publishing a manifest.

DefaultRemoteLogStorage wraps InterruptedException in RemoteStorageException, while copyPlannedSegment() treats it as an ordinary copy failure and returns false.

The task may then continue to write and commit a manifest containing an already-copied prefix, migration, expiration, or GC changes. In addition, RemoteLogManifestCommitter.commit() catches InterruptedException through its broad Exception handler and continues reconciliation/retries.

[swuferhong]

orphan sweep currently runs on every per-bucket tiering execution. With the default one-minute task interval, each bucket performs full listings of segment objects and manifest snapshots, potentially followed by one ZK read per deletion candidate and an unbounded number of deletions.

At scale, this can generate substantial load on ZooKeeper, occupy the remote-log scheduler threads, and delay normal tiering and TTL processing.

Please move orphan collection out of the per-minute tiering hot path and run it as a lower-frequency, independently scheduled GC process. The sweep should also be bounded, for example by limiting the number of objects scanned/deleted per run and supporting incremental progress with a cursor. ZK handle validation should preferably be performed per batch rather than once per object.