[server] Support RemoteLogManifest to handle overlap RemoteLogSegments

[gyang94]

Purpose

Remote log segments created by different replica leaders can have different physical boundaries. After a leader switch, a newly sealed local segment may overlap data that is already stored remotely. The previous tiering flow assumed physically non-overlapping segments, so it could not continue copying in this case.

This change allows physical remote log objects to overlap while exposing a single continuous, non-overlapping logical log view.

Closes #3835.

Brief change log

• Add logical start/end offsets to RemoteLogSegment while retaining physical offsets for remote object and index access.
• Merge copied segments into the existing manifest by clipping or replacing older logical ranges, and reject additions that introduce a gap.
• Track highestCopiedEndOffset independently from the current logical remote-log end so expiration or replacement does not lose tiering progress.
• Make offset lookup, timestamp lookup, and KV remote-log recovery honor logical segment bounds.
• Keep FetchLog v0 compatible by returning the maximal physically contiguous prefix before an overlap; the client resumes from its advanced offset in the next fetch.
• Remove superseded physical segment objects only after the updated manifest is committed.

Tests

• Added manifest serde compatibility and overlap merge tests, including clipping, replacement, multiple additions, gaps, and retained copy progress.
• Added remote tablet tests for logical lookup and FetchLog v0 overlap handling.
• Added remote KV fetcher coverage for segments produced with different replica boundaries.
• Added internal RPC compatibility tests for highestCopiedEndOffset.

API and Format

• No public client API changes.
• The remote manifest remains at version 1. Logical bounds and highest_copied_end_offset are optional fields and are omitted when they equal their legacy defaults.
• Existing manifests default logical bounds to physical bounds and copy progress to the remote-log end offset.
• The internal commit/notification RPCs add optional highest_copied_end_offset fields with the same legacy fallback.

Documentation

No user-facing configuration or API changes.

[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.

[wuchong]

Hi @gyang94 , I have a high-level concern about the current design. The implementation has become very complex and introduces a substantial amount of code that does not appear to be strictly necessary for the core purpose of this PR. In particular, it adds parallel Manifest V1/V2 paths, migration logic, a dedicated writer gate, a separate logical-reference model, and a replacement planner. Some of these paths duplicate existing manifest update and read logic.

I strongly suggest simplifying the design before merging this PR. Otherwise, future changes to offset semantics, fetching, expiration, GC, and compatibility will need to be kept consistent across multiple representations and code paths, resulting in a high long-term maintenance cost. Changes that are useful but unrelated to supporting overlapping segments should preferably be moved to separate follow-up PRs.

More specifically, I suggest the following:

1. Do not introduce remote.log.manifest-v2-writer-enabled.

   For rolling-upgrade compatibility, I think we should expose a switch for rolling the active segment instead and keep it disabled by default. During the rolling upgrade, active-segment rolling remains disabled, so no overlapping segments can be produced.

   We should also avoid bumping the manifest version. The existing manifest format can be extended with optional fields such as highestCopiedEndOffset. While active-segment rolling is disabled, highestCopiedEndOffset is always equal to remoteEndOffset, so old servers can continue reading manifests written by new servers while safely ignoring the additional field. Once every server has been upgraded, active-segment rolling can be enabled.

   This approach should allow us to remove a large amount of Manifest V2-specific code, including the writer gate, the V1/V2 dispatch paths, and RemoteLogManifestV2Migration.

2. Make RemoteLogSegment self-contained by storing its logical range directly.

   I suggest adding logicalStartOffset and logicalEndOffset to RemoteLogSegment itself. Separating logical bounds into RemoteLogSegmentReference means that any caller holding only a RemoteLogSegment can accidentally use the physical offsets, which is especially error-prone when a segment has been logically clipped because of an overlap.

   Keeping both the physical and logical ranges in the segment object would make its semantics explicit at every call site. It should also allow us to remove RemoteLogSegmentReference and the additional V2-specific read paths introduced around it, such as RemoteLogTablet#relevantRemoteLogSegmentReferences.

3. Reuse RemoteLogManifest#trimAndMerge instead of introducing RemoteLogManifestReplacementPlanner.

   I do not think we need a separate planner and another manifest-update state machine. We should extend RemoteLogManifest#trimAndMerge(deletedSegments, addedSegments) so that it can trim and merge overlapping segments correctly.

   Rolling the active segment can introduce overlaps, but it should not introduce gaps or skipped segments. Therefore, the additional gap/skip planning states do not seem necessary for the problem this PR is trying to solve. Handling overlap directly in trimAndMerge would keep manifest mutation in one place and avoid maintaining duplicate update rules.

The core design principle should be to avoid introducing a separate V2 path and instead adapt the existing path in place. This PR should focus exclusively on supporting overlapping segments. It should deliberately leave other pre-existing problems—such as log gaps, skipped logs, manifest generation, ZooKeeper CAS, and VersionedRemoteLogManifestHandle—out of scope. Those concerns can be addressed independently in follow-up PRs.

[wuchong]

Could we rename this option to remote.log.delete-delay and the constant to REMOTE_LOG_DELETE_DELAY? GC is implementation jargon that many operators may not recognize, while manifest-v2 exposes an internal format version in a public configuration key and will age poorly as the format evolves. This option delays physical deletion of unreferenced segments, obsolete manifest snapshots, and orphan objects, so delete-delay states the user-visible behavior more directly and remains version-neutral.

[wuchong]

We should allow remoteLogEndOffset to be updated back to -1, the same to remoteLogStartOffset. Otherwise, there is no difference between remoteLogEndOffset and highestCopiedEndOffset here.

[wuchong]

We shouldn't delete remote log segment before committing manifest, because commit may fail. In V1, we only delete expired segements after committing.

[wuchong]

We don't need to introduce another metric for failed delete remote log segment. This should already covered by metricGroup.remoteLogDeleteErrors().

[wuchong]

For large tables with tens of thousands of segments per bucket, this approach may lead to performance regression. I suggest using remoteLogTablet.findSegmentByTimestamp to optimize this method (note that findSegmentByTimestamp is not currently used in the PR). However, we must first correct its implementation to ensure semantic accuracy, as the presence of a logical end offset means the maximum timestamp may no longer be reliable. Please refer to my comments for further details.

Updates to remoteLogTablet.findSegmentByTimestamp:
Considering we already have logical start and logical end offset in RemoteLogSegment. We should retrieve the segment with the smallest logical start offset. If the logical end offset differs from the physical end offset (we can have a isEndOffsetClipped method on RemoteLogSegment), it indicates that the segment has been truncated, and the max timestamp may be inaccurate. In this case, also fetch the segment with the next timestamp and return both to the caller, allowing the caller to determine the correct segment. This should be implemented as a loop that continues until a non-truncated segment is found. Consequently, the return type should be changed to a List, sorted by logical start offset.

[wuchong]

we don't need version if we still use V1, and all other new fields are optional.

[wuchong]

we don't need generation as this is not the case we need to resolve in this PR.

[wuchong]

we don't need activeReferences if RemoteLogSegment itself contains logical start/end offset.

[wuchong]

We don't need (pre-compute) logicalRemoteLogStartOffset, and logicalRemoteLogEndOffset as they can be calculated by the logical start/end offset from first/last segemnt with O(1) complexity.

[wuchong]

I think we also don't need this. As we have the logical start offset in remoteLogSegmentList.

[wuchong]

Thanks @gyang94 for the updating. I think the overall code is very good now. I have some minor improvements and will open a follow-up PR for this. I will merge this PR for now.

[wuchong]

LGTM.