Skip to content

[server] Update remote log offsets before local segment cleanup #4254

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wuchong merged 2 commits into from
Sep 10, 2026
Merged

[server] Update remote log offsets before local segment cleanup #4254

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a496a31
[server] Update remote log offsets before local segment cleanup
Sep 8, 2026
1e9002b
modified based on CR
Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 44 additions & 15 deletions fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -638,36 +638,54 @@ public long lookupOffsetForTimestamp(long startTimestamp) throws IOException {
return findOffset;
}

public void updateRemoteLogStartOffset(long remoteLogStartOffset) {
private void updateRemoteLogStartOffset(long remoteLogStartOffset) {
long prev = this.remoteLogStartOffset;
if (prev == Long.MAX_VALUE || remoteLogStartOffset > prev) {
this.remoteLogStartOffset = remoteLogStartOffset;
}
}

/** Updates the size of the log segments currently retained in remote storage. */
public void updateRemoteLogSize(long remoteLogSize) {
this.remoteLogSize = remoteLogSize;
}

public void updateRemoteLogEndOffset(long remoteLogEndOffset) {
/**
* Updates the remote log offsets from one committed manifest.
*
* <p>The remote-readable start and end offsets are published before advancing the copied
* watermark and deleting local segments. This prevents fetches from observing locally deleted
* offsets before the corresponding remote range becomes readable. Local segments are cleaned up
* at most once.
*/
public void updateRemoteLogOffsets(
long newRemoteLogStartOffset, long remoteLogEndOffset, long highestCopiedEndOffset) {
updateRemoteLogStartOffset(newRemoteLogStartOffset);

boolean shouldCleanup = false;
if ((remoteLogEndOffset == -1L && this.remoteLogEndOffset != -1L)
|| remoteLogEndOffset > this.remoteLogEndOffset) {
this.remoteLogEndOffset = remoteLogEndOffset;
// Before highestCopiedEndOffset was introduced, remoteLogEndOffset was also the copy
// progress watermark. Preserve that behavior for existing callers.
if (remoteLogEndOffset >= 0L) {
this.highestCopiedEndOffset =
Math.max(this.highestCopiedEndOffset, remoteLogEndOffset);
}

// try to delete these segments already exist in remote storage.
deleteSegmentsAlreadyExistsInRemote();
shouldCleanup = true;
}
}

public void updateHighestCopiedEndOffset(long highestCopiedEndOffset) {
if (highestCopiedEndOffset > this.highestCopiedEndOffset) {
this.highestCopiedEndOffset = highestCopiedEndOffset;
shouldCleanup = true;
}
// The remote-readable end offset should never trail the copied watermark unless the
// manifest is empty (remoteLogEndOffset == -1). A non-empty manifest with a readable end
// behind the copied watermark means local segments could be cleaned up beyond the range
// that is actually readable from remote, which risks an unreadable offset gap.
if (this.remoteLogEndOffset != -1L
&& this.remoteLogEndOffset < this.highestCopiedEndOffset) {
LOG.warn(
"Remote readable end offset {} is behind copied watermark {} for bucket {}; "
+ "local cleanup will be bounded by the readable end offset.",
this.remoteLogEndOffset,
this.highestCopiedEndOffset,
getTableBucket());
}
if (shouldCleanup) {
deleteSegmentsAlreadyExistsInRemote();
}
}
Expand Down Expand Up @@ -789,8 +807,19 @@ public void loadWriterSnapshot(long lastOffset) throws IOException {
}
}

/**
* Deletes eligible local segments that have already been copied to remote storage.
*
* <p>For a non-empty manifest, cleanup never advances past the remote-readable end offset. An
* empty manifest keeps using the copied watermark so retention can continue after all remote
* segments have expired.
*/
public void deleteSegmentsAlreadyExistsInRemote() {
cleanupSegments(highestCopiedEndOffset, this::cleanupTieredSegments);
long cleanupToOffset =
remoteLogEndOffset == -1L
? highestCopiedEndOffset
: Math.min(remoteLogEndOffset, highestCopiedEndOffset);
cleanupSegments(cleanupToOffset, this::cleanupTieredSegments);
}

/**
Expand Down
8 changes: 4 additions & 4 deletions fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -400,11 +400,11 @@ private boolean tryToCommitRemoteLogManifest(
// TODO: commit with version to avoid the manifest has been updated
remoteLogTablet.loadRemoteLogManifest(newRemoteLogManifest);
LogTablet logTablet = replica.getLogTablet();
logTablet.updateRemoteLogStartOffset(newRemoteLogStartOffset);
logTablet.updateHighestCopiedEndOffset(

logTablet.updateRemoteLogOffsets(
newRemoteLogStartOffset,
newRemoteLogEndOffset,
newRemoteLogManifest.getHighestCopiedEndOffset());
// make the local log cleaner clean log segments that are committed to remote.
logTablet.updateRemoteLogEndOffset(newRemoteLogEndOffset);
logTablet.updateRemoteLogSize(newRemoteLogSize);
return true;
}
Expand Down
8 changes: 5 additions & 3 deletions fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,9 +163,11 @@ public void registerReplica(Replica replica) throws Exception {
remoteLogManifestHandleOpt.get().getRemoteLogManifestPath());
remoteLog.loadRemoteLogManifest(manifest);
}
log.updateHighestCopiedEndOffset(remoteLog.getHighestCopiedEndOffset());
log.updateRemoteLogEndOffset(remoteLog.getRemoteLogEndOffset().orElse(-1L));
log.updateRemoteLogStartOffset(remoteLog.getRemoteLogStartOffset());

log.updateRemoteLogOffsets(
remoteLog.getRemoteLogStartOffset(),
remoteLog.getRemoteLogEndOffset().orElse(-1L),
remoteLog.getHighestCopiedEndOffset());
log.updateRemoteLogSize(remoteLog.getRemoteSizeInBytes());
// leader needs to register the remote log metrics
remoteLog.registerMetrics(replica.bucketMetrics());
Expand Down
8 changes: 3 additions & 5 deletions fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1347,12 +1347,10 @@ public void notifyRemoteLogOffsets(
// remote.
TableBucket tb = notifyRemoteLogOffsetsData.getTableBucket();
LogTablet logTablet = getReplicaOrException(tb).getLogTablet();
logTablet.updateHighestCopiedEndOffset(
logTablet.updateRemoteLogOffsets(
notifyRemoteLogOffsetsData.getRemoteLogStartOffset(),
notifyRemoteLogOffsetsData.getRemoteLogEndOffset(),
notifyRemoteLogOffsetsData.getHighestCopiedEndOffset());
logTablet.updateRemoteLogStartOffset(
notifyRemoteLogOffsetsData.getRemoteLogStartOffset());
logTablet.updateRemoteLogEndOffset(
notifyRemoteLogOffsetsData.getRemoteLogEndOffset());
responseCallback.accept(new NotifyRemoteLogOffsetsResponse());
});
}
Expand Down
15 changes: 9 additions & 6 deletions fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,17 +115,20 @@ public void teardown() throws Exception {
}

@Test
void testRemoteLogEndOffsetCanReset() {
logTablet.updateRemoteLogStartOffset(0L);
logTablet.updateRemoteLogEndOffset(10L);
void testRemoteLogOffsetsCanResetAfterEmptyManifest() {
logTablet.updateRemoteLogOffsets(0L, 10L, 10L);
assertThat(logTablet.canFetchFromRemoteLog(0L)).isTrue();
assertThat(logTablet.canFetchFromRemoteLog(10L)).isFalse();

logTablet.updateRemoteLogEndOffset(-1L);
logTablet.updateRemoteLogOffsets(Long.MAX_VALUE, -1L, 10L);
assertThat(logTablet.canFetchFromRemoteLog(0L)).isFalse();
assertThat(logTablet.canFetchFromRemoteLog(10L)).isFalse();

// A new non-empty range can become readable after the empty state.
logTablet.updateRemoteLogEndOffset(5L);
assertThat(logTablet.canFetchFromRemoteLog(0L)).isTrue();
logTablet.updateRemoteLogOffsets(10L, 20L, 20L);
assertThat(logTablet.canFetchFromRemoteLog(0L)).isFalse();
assertThat(logTablet.canFetchFromRemoteLog(10L)).isTrue();
assertThat(logTablet.canFetchFromRemoteLog(20L)).isFalse();
}

@Test
Expand Down
66 changes: 62 additions & 4 deletions fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.remote.RemoteLogFetchInfo;
import org.apache.fluss.remote.RemoteLogManifest;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.protocol.ApiError;
Expand All @@ -48,6 +49,7 @@

import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
Expand Down Expand Up @@ -316,6 +318,48 @@ void testCommitDeleteLogSegmentsFromRemoteFailed2(boolean partitionTable) throws
.collect(Collectors.toSet()));
}

@Test
void testFetchRemainsAvailableWhenRemoteOffsetsAdvance() throws Exception {
TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0);
makeLogTableAsLeader(tableBucket, false);
LogTablet logTablet = replicaManager.getReplicaOrException(tableBucket).getLogTablet();

// Local segments are [0, 10), [10, 20), [20, 30), [30, 40), and [40, 50).
addMultiSegmentsToLogTablet(logTablet, 5);

List<RemoteLogSegment> remoteSegments = new ArrayList<>();
for (int i = 0; i < 4; i++) {
remoteSegments.add(copyLogSegmentToRemote(logTablet, remoteLogStorage, i));
}
RemoteLogTablet remoteLogTablet = remoteLogManager.remoteLogTablet(tableBucket);
remoteLogTablet.loadRemoteLogManifest(
new RemoteLogManifest(
logTablet.getPhysicalTablePath(),
tableBucket,
remoteSegments.subList(0, 2),
40L));

// An older manifest is readable only up to 20 even though copying has advanced to 40.
// Cleanup must remain bounded by the readable end, leaving offset 25 available locally.
logTablet.updateRemoteLogOffsets(0L, 20L, 40L);
assertThat(logTablet.localLogStartOffset()).isEqualTo(20L);

FetchLogResultForBucket localResult = fetch(tableBucket, 25L);
assertThat(localResult.getError()).isEqualTo(ApiError.NONE);
assertThat(localResult.fetchFromRemote()).isFalse();

remoteLogTablet.loadRemoteLogManifest(
new RemoteLogManifest(
logTablet.getPhysicalTablePath(), tableBucket, remoteSegments, 40L));
logTablet.updateRemoteLogOffsets(0L, 40L, 40L);
assertThat(logTablet.localLogStartOffset()).isEqualTo(30L);
assertThat(logTablet.canFetchFromRemoteLog(25L)).isTrue();

@gyang94 gyang94 Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here we checked the final state after updateRemoteLogEndOffset. It can't check whether we have the correct update processing order. We must 1) update copied watermark first then 2) delete local segment.
The problem is even if we first do operation 2 then 1, this test will also pass.

@loserwang1024 loserwang1024 Sep 9, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

t's hard to test this unless we add a hook before cleanupSegments, which is too intrusive to the production code.

@gyang94 gyang94 Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure. It is not a blocker.


FetchLogResultForBucket remoteResult = fetch(tableBucket, 25L);
assertThat(remoteResult.getError()).isEqualTo(ApiError.NONE);
assertThat(remoteResult.fetchFromRemote()).isTrue();
}

@ParameterizedTest
@ValueSource(booleans = {true, false})
void testFetchRecordsFromRemote(boolean partitionTable) throws Exception {
Expand All @@ -334,7 +378,7 @@ void testFetchRecordsFromRemote(boolean partitionTable) throws Exception {

// 1. first, fetch records from remote.
// mock to update remote log end offset and delete local log segments.
logTablet.updateRemoteLogEndOffset(40L);
logTablet.updateRemoteLogOffsets(0L, 40L, 40L);
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> future =
new CompletableFuture<>();
replicaManager.fetchLogRecords(
Expand Down Expand Up @@ -380,7 +424,7 @@ void testRemoteFirstFetchPrefersRemoteWhenLocalStillHasRecords(boolean partition
LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
addMultiSegmentsToLogTablet(logTablet, 5);
remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
logTablet.updateRemoteLogEndOffset(40L);
logTablet.updateRemoteLogOffsets(0L, 40L, 40L);

Map<TableBucket, FetchReqInfo> fetchData =
Collections.singletonMap(tb, new FetchReqInfo(tb.getTableId(), 35L, 1024 * 1024));
Expand Down Expand Up @@ -427,7 +471,7 @@ void testRemoteFirstFetchRejectsNonLeader(boolean partitionTable) throws Excepti
LogTablet logTablet = replica.getLogTablet();
addMultiSegmentsToLogTablet(logTablet, 5);
remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
logTablet.updateRemoteLogEndOffset(40L);
logTablet.updateRemoteLogOffsets(0L, 40L, 40L);

int newLeaderId = TABLET_SERVER_ID + 1;
replica.makeFollower(
Expand Down Expand Up @@ -480,7 +524,7 @@ void testCleanupLocalSegments(boolean partitionTable) throws Exception {
assertThat(remoteLog.allRemoteLogSegments()).hasSize(4);

// 3. mock to update remote end offset, shouldn't cleanup local segments
logTablet.updateRemoteLogEndOffset(40L);
logTablet.updateRemoteLogOffsets(0L, 40L, 40L);
assertThat(logTablet.getSegments()).hasSize(5);

// 4. mock to update min retain, should remove the first 3 segments (end offset < 33)
Expand Down Expand Up @@ -853,6 +897,20 @@ private TableBucket makeTableBucket(boolean partitionTable) {
return makeTableBucket(DATA1_TABLE_ID, partitionTable);
}

private FetchLogResultForBucket fetch(TableBucket tableBucket, long fetchOffset)
throws Exception {
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> future =
new CompletableFuture<>();
replicaManager.fetchLogRecords(
new FetchParams(-1, Integer.MAX_VALUE),
Collections.singletonMap(
tableBucket,
new FetchReqInfo(tableBucket.getTableId(), fetchOffset, 1024 * 1024)),
null,
future::complete);
return future.get().get(tableBucket);
}

private TableBucket makeTableBucket(long tableId, boolean partitionTable) {
if (partitionTable) {
return new TableBucket(tableId, 0L, 0);
Expand Down
8 changes: 3 additions & 5 deletions fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,11 +143,9 @@ void testRemoteLogTTL(boolean partitionTable) throws Exception {
assertThat(remoteLog.getRemoteLogStartOffset()).isEqualTo(Long.MAX_VALUE);
assertThat(remoteLog.getHighestCopiedEndOffset()).isEqualTo(40L);

// Fetch records from remote.
// mock to update remote log end offset and remote log start offset as
// NotifyRemoteLogOffsetsRequest do.
logTablet.updateRemoteLogStartOffset(40L);
logTablet.updateRemoteLogEndOffset(40L);
// Fetch records from remote. Mock the empty manifest state propagated by
// NotifyRemoteLogOffsetsRequest.
logTablet.updateRemoteLogOffsets(Long.MAX_VALUE, -1L, 40L);
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> future =
new CompletableFuture<>();
replicaManager.fetchLogRecords(
Expand Down
7 changes: 4 additions & 3 deletions fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,7 @@ void testExpiredActiveSegmentWaitsForHighWatermark(boolean partitionTable) throw
LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();

addMultiSegmentsToLogTablet(logTablet, 5);
logTablet.updateHighestCopiedEndOffset(40L);
logTablet.updateRemoteLogOffsets(Long.MAX_VALUE, -1L, 40L);
manualClock.advanceTime(Duration.ofMinutes(90));
logTablet.updateHighWatermark(logTablet.localLogEndOffset() - 1L);
logManager.cleanupExpiredLocalLogSegments();
Expand Down Expand Up @@ -191,7 +191,8 @@ void testTtlCleanupBoundedByHighestCopiedEndOffset(boolean partitionTable) throw

addMultiSegmentsToLogTablet(logTablet, 5);
updateTableConfig(replica, ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, "5");
logTablet.updateHighestCopiedEndOffset(20L);
logTablet.updateRemoteLogOffsets(Long.MAX_VALUE, -1L, 20L);
assertThat(logTablet.canFetchFromRemoteLog(0L)).isFalse();

manualClock.advanceTime(Duration.ofMinutes(90));
logManager.cleanupExpiredLocalLogSegments();
Expand All @@ -200,7 +201,7 @@ void testTtlCleanupBoundedByHighestCopiedEndOffset(boolean partitionTable) throw
assertThat(logTablet.localLogStartOffset()).isEqualTo(20L);
assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L);

logTablet.updateRemoteLogEndOffset(40L);
logTablet.updateRemoteLogOffsets(0L, 40L, 40L);
logManager.cleanupExpiredLocalLogSegments();

assertThat(logTablet.getSegments()).hasSize(2);
Expand Down
2 changes: 1 addition & 1 deletion ...erver/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,7 @@ void testRestoreKvMinRetainOffsetFromDelayedEmptyFetchResponse() throws Exceptio
leaderReplica.getLogTablet().updateHighWatermark(30L);
followerReplica.getLogTablet().updateHighWatermark(30L);
leaderReplica.getLogTablet().updateMinRetainOffset(30L);
followerReplica.getLogTablet().updateHighestCopiedEndOffset(30L);
followerReplica.getLogTablet().updateRemoteLogOffsets(Long.MAX_VALUE, -1L, 30L);

assertThat(leaderReplica.getLocalLogEndOffset()).isEqualTo(30L);
assertThat(followerReplica.getLocalLogEndOffset()).isEqualTo(30L);
Expand Down
Loading