[server] Fix ReplicaFetcherThread keeps throwing UnknownServerException because of corrupt index file

Author: LiebingYu

fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java

@@ -176,7 +176,7 @@ public class ReplicaManager {
     private final SnapshotContext kvSnapshotContext;
 
     // remote log manager for remote log storage.
-    private final RemoteLogManager remoteLogManager;
+    protected final RemoteLogManager remoteLogManager;
 
     // for metrics
     private final TabletServerMetricGroup serverMetricGroup;
@@ -217,7 +217,7 @@ public ReplicaManager(
     }
 
     @VisibleForTesting
-    ReplicaManager(
+    protected ReplicaManager(
             Configuration conf,
             Scheduler scheduler,
             LogManager logManager,
@@ -1101,7 +1101,9 @@ public Map<TableBucket, LogReadResult> readFromLog(
 
                 FetchLogResultForBucket result;
                 if (replica != null && e instanceof LogOffsetOutOfRangeException) {
-                    result = handleFetchOutOfRangeException(replica, fetchOffset, e);
+                    result =
+                            handleFetchOutOfRangeException(
+                                    replica, fetchOffset, fetchParams.isFromFollower(), e);
                 } else {
                     result = new FetchLogResultForBucket(tb, ApiError.fromThrowable(e));
                 }
@@ -1113,7 +1115,7 @@ public Map<TableBucket, LogReadResult> readFromLog(
     }
 
     private FetchLogResultForBucket handleFetchOutOfRangeException(
-            Replica replica, long fetchOffset, Exception e) {
+            Replica replica, long fetchOffset, boolean isFromFollower, Exception e) {
         TableBucket tb = replica.getTableBucket();
         if (fetchOffset == FetchParams.FETCH_FROM_EARLIEST_OFFSET) {
             fetchOffset = replica.getLogStartOffset();
@@ -1131,7 +1133,8 @@ private FetchLogResultForBucket handleFetchOutOfRangeException(
         // of RemoteLogSegment. For client fetcher, it will fetch the log from remote in client.
         // For follower, it can update its local metadata to adjust the next fetch offset.
         else if (canFetchFromRemoteLog(replica, fetchOffset)) {
-            RemoteLogFetchInfo remoteLogFetchInfo = fetchLogFromRemote(replica, fetchOffset);
+            RemoteLogFetchInfo remoteLogFetchInfo =
+                    fetchLogFromRemote(replica, fetchOffset, isFromFollower);
             if (remoteLogFetchInfo != null) {
                 return new FetchLogResultForBucket(
                         tb, remoteLogFetchInfo, replica.getLogHighWatermark());
@@ -1157,13 +1160,20 @@ private boolean canFetchFromRemoteLog(Replica replica, long fetchOffset) {
         return replica.getLogTablet().canFetchFromRemoteLog(fetchOffset);
     }
 
-    private @Nullable RemoteLogFetchInfo fetchLogFromRemote(Replica replica, long fetchOffset) {
+    private @Nullable RemoteLogFetchInfo fetchLogFromRemote(
+            Replica replica, long fetchOffset, boolean isFromFollower) {
         List<RemoteLogSegment> remoteLogSegmentList =
                 remoteLogManager.relevantRemoteLogSegments(replica.getTableBucket(), fetchOffset);
         if (!remoteLogSegmentList.isEmpty()) {
-            int firstStartPos =
-                    remoteLogManager.lookupPositionForOffset(
-                            remoteLogSegmentList.get(0), fetchOffset);
+            // follower does not require firstStartPos, so we do not look up this value.
+            // On one hand, this reduces query overhead, and on the other hand,
+            // it helps prevent unexpected issues caused by corrupted index files.
+            int firstStartPos = -1;
+            if (!isFromFollower) {
+                firstStartPos =
+                        remoteLogManager.lookupPositionForOffset(
+                                remoteLogSegmentList.get(0), fetchOffset);
+            }
             PhysicalTablePath physicalTablePath = replica.getPhysicalTablePath();
             FsPath remoteLogTabletDir =
                     FlussPaths.remoteLogTabletDir(

fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java

@@ -26,6 +26,7 @@
 import org.apache.fluss.rpc.protocol.ApiError;
 import org.apache.fluss.server.coordinator.TestCoordinatorGateway;
 import org.apache.fluss.server.entity.FetchReqInfo;
+import org.apache.fluss.server.exception.CorruptIndexException;
 import org.apache.fluss.server.log.FetchParams;
 import org.apache.fluss.server.log.LogTablet;
 import org.apache.fluss.server.replica.Replica;
@@ -34,6 +35,10 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
@@ -351,6 +356,62 @@ void testFetchRecordsFromRemote(boolean partitionTable) throws Exception {
         assertThat(resultForBucket.fetchFromRemote()).isFalse();
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testFetchRecordsFromRemoteWithCorruptIndex(boolean partitionTable) throws Exception {
+        TableBucket tb = makeTableBucket(partitionTable);
+        // Need to make leader by ReplicaManager.
+        makeLogTableAsLeader(tb, partitionTable);
+        LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
+        addMultiSegmentsToLogTablet(logTablet, 5);
+
+        File file = logTablet.getSegments().get(0).timeIndex().file();
+        // mock corrupt index
+        try (FileChannel fileChannel = FileChannel.open(file.toPath(), StandardOpenOption.APPEND)) {
+            for (int i = 0; i < 12; i++) {
+                fileChannel.write(ByteBuffer.wrap(new byte[] {0}));
+            }
+        }
+
+        // trigger RLMTask copy local log segment to remote and update metadata.
+        remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+        List<RemoteLogSegment> remoteLogSegmentList =
+                remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+        assertThat(remoteLogSegmentList.size()).isEqualTo(4);
+
+        logTablet.updateRemoteLogEndOffset(40L);
+
+        // 1. fetch log records from client, should throw CorruptIndexException since we mock a
+        // corrupt index
+        CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> future =
+                new CompletableFuture<>();
+        CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> finalFuture = future;
+        assertThatThrownBy(
+                        () ->
+                                replicaManager.fetchLogRecords(
+                                        new FetchParams(-1, Integer.MAX_VALUE),
+                                        Collections.singletonMap(
+                                                tb,
+                                                new FetchReqInfo(tb.getTableId(), 0L, 1024 * 1024)),
+                                        finalFuture::complete))
+                .isInstanceOf(CorruptIndexException.class);
+
+        // 2. fetch log records from follower, should succeed since we don't need to lookup index
+        future = new CompletableFuture<>();
+        replicaManager.fetchLogRecords(
+                new FetchParams(1, Integer.MAX_VALUE),
+                Collections.singletonMap(tb, new FetchReqInfo(tb.getTableId(), 0L, 1024 * 1024)),
+                future::complete);
+        Map<TableBucket, FetchLogResultForBucket> result = future.get();
+        assertThat(result.size()).isEqualTo(1);
+        FetchLogResultForBucket resultForBucket = result.get(tb);
+        assertThat(resultForBucket.getError()).isEqualTo(ApiError.NONE);
+        assertThat(resultForBucket.getTableBucket()).isEqualTo(tb);
+        assertThat(resultForBucket.fetchFromRemote()).isTrue();
+        assertThat(resultForBucket.records()).isNull();
+        assertThat(resultForBucket.getHighWatermark()).isEqualTo(50L);
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     void testCleanupLocalSegments(boolean partitionTable) throws Exception {