Skip to content

[server] release remote log index cache entries on deletion #3737

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

Open
zuston wants to merge 1 commit into
base: main
Choose a base branch
from
+45 −0
Open

[server] release remote log index cache entries on deletion #3737

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 9 additions & 0 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 @@ -43,6 +43,7 @@
import java.util.Objects;
import java.util.OptionalLong;
import java.util.UUID;
import java.util.stream.Collectors;

import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeCommitRemoteLogManifestRequest;

Expand All @@ -57,6 +58,7 @@ public class LogTieringTask implements Runnable {
private final PhysicalTablePath physicalTablePath;
private final TableBucket tableBucket;
private final RemoteLogStorage remoteLogStorage;
private final RemoteLogIndexCache remoteLogIndexCache;
private final CoordinatorGateway coordinatorGateway;
private final Clock clock;
private final int maxUploadSegmentsPerTask;
Expand All @@ -71,6 +73,7 @@ public LogTieringTask(
Replica replica,
RemoteLogTablet remoteLog,
RemoteLogStorage remoteLogStorage,
RemoteLogIndexCache remoteLogIndexCache,
CoordinatorGateway coordinatorGateway,
Clock clock,
int maxUploadSegmentsPerTask) {
Expand All @@ -79,6 +82,7 @@ public LogTieringTask(
this.physicalTablePath = replica.getPhysicalTablePath();
this.tableBucket = replica.getTableBucket();
this.remoteLogStorage = remoteLogStorage;
this.remoteLogIndexCache = remoteLogIndexCache;
this.coordinatorGateway = coordinatorGateway;
this.clock = clock;
this.maxUploadSegmentsPerTask = maxUploadSegmentsPerTask;
Expand Down Expand Up @@ -154,6 +158,11 @@ private void runOnce() throws InterruptedException {

if (success) {
if (!expiredRemoteLogSegments.isEmpty()) {
// Release the mmap-backed local indexes before deleting the remote files.
remoteLogIndexCache.removeAll(

@fresh-borzoni fresh-borzoni Jul 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Q: shall we removeAll after deleteRemoteLogSegmentFiles instead of before?
In the gap between manifest commit and remote deletion, an in-flight reader can miss the cache and re-download the index (remote files still exist at that point), re-inserting an entry that nothing will ever remove again, essentially the exact leak this PR fixes.

The read-refcount TODO would fix the reader-error half of this race, but not the leak half and the reorder(the thing that I propose here) composes with it if it ever lands: drain -> delete files -> removeAll.

WDYT?

expiredRemoteLogSegments.stream()
.map(RemoteLogSegment::remoteLogSegmentId)
.collect(Collectors.toList()));
// 3. For these expiredRemoteLogSegments, we will delete remote log
// segment files from remote after commit the remote log manifest.
// TODO introduce the read reference count to avoid deleting remote log
Expand Down
1 change: 1 addition & 0 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 @@ -315,6 +315,7 @@ private void doHandleLeaderReplica(
replica,
remoteLog,
remoteLogStorage,
remoteLogIndexCache(replica.getLogTablet().getDataDir()),
coordinatorGateway,
clock,
maxUploadSegmentsPerTask);
Expand Down
35 changes: 35 additions & 0 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 @@ -18,11 +18,13 @@
package org.apache.fluss.server.log.remote;

import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.protocol.Errors;
import org.apache.fluss.server.entity.FetchReqInfo;
import org.apache.fluss.server.log.FetchParams;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.log.remote.RemoteLogIndexCache.Entry;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.params.ParameterizedTest;
Expand Down Expand Up @@ -69,6 +71,31 @@ void testRemoteLogTTL(boolean partitionTable) throws Exception {
assertThat(remoteLog.getRemoteLogStartOffset()).isEqualTo(0L);
assertThat(remoteLog.getRemoteLogEndOffset()).hasValue(40L);

// Materialize the index cache entries for all remote log segments.
RemoteLogIndexCache indexCache =
remoteLogManager.getRemoteLogIndexCache(logTablet.getDataDir());
for (RemoteLogSegment remoteLogSegment : remoteLog.allRemoteLogSegments()) {
remoteLogManager.lookupPositionForOffset(
remoteLogSegment, remoteLogSegment.remoteLogStartOffset());
}
assertThat(indexCache.getInternalCache().asMap()).hasSize(4);
RemoteLogSegment expiredSegment =
remoteLog.allRemoteLogSegments().stream()
.filter(segment -> segment.remoteLogStartOffset() < 20L)
.findFirst()
.get();
RemoteLogSegment retainedSegment =
remoteLog.allRemoteLogSegments().stream()
.filter(segment -> segment.remoteLogStartOffset() >= 20L)
.findFirst()
.get();
Entry expiredIndexEntry =
indexCache.getInternalCache().getIfPresent(expiredSegment.remoteLogSegmentId());
Entry retainedIndexEntry =
indexCache.getInternalCache().getIfPresent(retainedSegment.remoteLogSegmentId());
assertThat(expiredIndexEntry).isNotNull();
assertThat(retainedIndexEntry).isNotNull();

// advance time past TTL (7 days)
manualClock.advanceTime(Duration.ofDays(7).plusHours(1));

Expand Down Expand Up @@ -96,6 +123,14 @@ void testRemoteLogTTL(boolean partitionTable) throws Exception {
segment ->
assertThat(segment.remoteLogStartOffset())
.isGreaterThanOrEqualTo(20L));
assertThat(indexCache.getInternalCache().asMap())
.hasSize(2)
.doesNotContainKeys(expiredSegment.remoteLogSegmentId())
.containsKey(retainedSegment.remoteLogSegmentId());
assertThat(expiredIndexEntry.offsetIndex().file()).doesNotExist();
assertThat(expiredIndexEntry.timeIndex().file()).doesNotExist();
assertThat(retainedIndexEntry.offsetIndex().file()).exists();
assertThat(retainedIndexEntry.timeIndex().file()).exists();

// now advance lake log end offset to include all remaining segments
logTablet.updateLakeLogEndOffset(40L);
Expand Down
Loading