[server] Fix orphan segments after full truncation

fluss-server/src/main/java/org/apache/fluss/server/log/LocalLog.java

@@ -324,19 +324,30 @@ void removeAndDeleteSegments(List<LogSegment> segmentsToDelete, SegmentDeletionR
     LogSegment createAndDeleteSegment(
             long newOffset, LogSegment segmentToDelete, SegmentDeletionReason reason)
             throws IOException {
-        // delete the old segment.
-        if (newOffset == segmentToDelete.getBaseOffset()) {
-            deleteSegmentFiles(Collections.singletonList(segmentToDelete), reason);
+        boolean replaceAtSameOffset = newOffset == segmentToDelete.getBaseOffset();
+        if (replaceAtSameOffset) {
+            segmentToDelete.changeFileSuffixes("", FlussPaths.DELETED_FILE_SUFFIX);
         }
-        reason.logReason(Collections.singletonList(segmentToDelete));
 
-        // open a new segment.
-        LogSegment newSegment = LogSegment.open(logTabletDir, newOffset, config, logFormat);
+        LogSegment newSegment;
+        try {
+            newSegment = LogSegment.open(logTabletDir, newOffset, config, logFormat);
+        } catch (IOException e) {
+            if (replaceAtSameOffset) {
+                try {
+                    segmentToDelete.changeFileSuffixes(FlussPaths.DELETED_FILE_SUFFIX, "");
+                } catch (IOException rollbackException) {
+                    e.addSuppressed(rollbackException);
+                }
+            }
+            throw e;
+        }
         segments.add(newSegment);
 
-        if (newOffset != segmentToDelete.getBaseOffset()) {
+        if (!replaceAtSameOffset) {
             segments.remove(segmentToDelete.getBaseOffset());
         }
+        deleteSegmentFiles(Collections.singletonList(segmentToDelete), reason);
         return newSegment;
     }
 

fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java

@@ -1301,7 +1301,12 @@ private LogAppendInfo analyzeAndValidateRecords(MemoryLogRecords records) {
                 }
 
                 // update write append info.
-                updateWriterAppendInfo(writerStateManager, batch, updatedWriters, isAppendAsLeader);
+                updateWriterAppendInfo(
+                        writerStateManager,
+                        batch,
+                        updatedWriters,
+                        isAppendAsLeader,
+                        WriterAppendInfo.SequenceValidation.ENFORCE);
             }
         }
 
@@ -1439,15 +1444,17 @@ private static void updateWriterAppendInfo(
             WriterStateManager writerStateManager,
             LogRecordBatch batch,
             Map<Long, WriterAppendInfo> writers,
-            boolean isAppendAsLeader) {
+            boolean isAppendAsLeader,
+            WriterAppendInfo.SequenceValidation sequenceValidation) {
         long writerId = batch.writerId();
         // update writers.
         WriterAppendInfo appendInfo =
                 writers.computeIfAbsent(writerId, id -> writerStateManager.prepareUpdate(writerId));
         appendInfo.append(
                 batch,
                 writerStateManager.isWriterInBatchExpired(System.currentTimeMillis(), batch),
-                isAppendAsLeader);
+                isAppendAsLeader,
+                sequenceValidation);
     }
 
     static void rebuildWriterState(
@@ -1565,7 +1572,14 @@ private static void loadWritersFromRecords(
         Map<Long, WriterAppendInfo> loadedWriters = new HashMap<>();
         for (LogRecordBatch batch : records.batches()) {
             if (batch.hasWriterId()) {
-                updateWriterAppendInfo(writerStateManager, batch, loadedWriters, false);
+                // The records have already been accepted and persisted. Recovery rebuilds writer
+                // state without applying online client sequence validation.
+                updateWriterAppendInfo(
+                        writerStateManager,
+                        batch,
+                        loadedWriters,
+                        false,
+                        WriterAppendInfo.SequenceValidation.WARN_AND_ACCEPT);
             }
         }
         loadedWriters.values().forEach(writerStateManager::update);

fluss-server/src/main/java/org/apache/fluss/server/log/WriterAppendInfo.java

@@ -21,13 +21,23 @@
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.record.LogRecordBatch;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import static org.apache.fluss.record.LogRecordBatchFormat.NO_BATCH_SEQUENCE;
 
 /**
  * This class is used to validate the records appended by a given writer before they are written to
  * log. It's initialized with writer's state after the last successful append.
  */
 public class WriterAppendInfo {
+    private static final Logger LOG = LoggerFactory.getLogger(WriterAppendInfo.class);
+
+    enum SequenceValidation {
+        ENFORCE,
+        WARN_AND_ACCEPT
+    }
+
     private final long writerId;
     private final TableBucket tableBucket;
     private final WriterStateEntry currentEntry;
@@ -46,14 +56,23 @@ public long writerId() {
 
     public void append(
             LogRecordBatch batch, boolean isWriterInBatchExpired, boolean isAppendAsLeader) {
+        append(batch, isWriterInBatchExpired, isAppendAsLeader, SequenceValidation.ENFORCE);
+    }
+
+    void append(
+            LogRecordBatch batch,
+            boolean isWriterInBatchExpired,
+            boolean isAppendAsLeader,
+            SequenceValidation sequenceValidation) {
         LogOffsetMetadata firstOffsetMetadata = new LogOffsetMetadata(batch.baseLogOffset());
         appendDataBatch(
                 batch.batchSequence(),
                 firstOffsetMetadata,
                 batch.lastLogOffset(),
                 isWriterInBatchExpired,
                 isAppendAsLeader,
-                batch.commitTimestamp());
+                batch.commitTimestamp(),
+                sequenceValidation);
     }
 
     public void appendDataBatch(
@@ -63,7 +82,38 @@ public void appendDataBatch(
             boolean isWriterInBatchExpired,
             boolean isAppendAsLeader,
             long batchTimestamp) {
-        maybeValidateDataBatch(batchSequence, isWriterInBatchExpired, lastOffset, isAppendAsLeader);
+        appendDataBatch(
+                batchSequence,
+                firstOffsetMetadata,
+                lastOffset,
+                isWriterInBatchExpired,
+                isAppendAsLeader,
+                batchTimestamp,
+                SequenceValidation.ENFORCE);
+    }
+
+    private void appendDataBatch(
+            int batchSequence,
+            LogOffsetMetadata firstOffsetMetadata,
+            long lastOffset,
+            boolean isWriterInBatchExpired,
+            boolean isAppendAsLeader,
+            long batchTimestamp,
+            SequenceValidation sequenceValidation) {
+        maybeValidateDataBatch(
+                batchSequence,
+                isWriterInBatchExpired,
+                lastOffset,
+                isAppendAsLeader,
+                sequenceValidation);
+        appendDataBatch(batchSequence, firstOffsetMetadata, lastOffset, batchTimestamp);
+    }
+
+    private void appendDataBatch(
+            int batchSequence,
+            LogOffsetMetadata firstOffsetMetadata,
+            long lastOffset,
+            long batchTimestamp) {
         updatedEntry.addBath(
                 batchSequence,
                 lastOffset,
@@ -75,21 +125,30 @@ private void maybeValidateDataBatch(
             int appendFirstSeq,
             boolean isWriterInBatchExpired,
             long lastOffset,
-            boolean isAppendAsLeader) {
-        int currentLastSeq =
-                !updatedEntry.isEmpty()
-                        ? updatedEntry.lastBatchSequence()
-                        : currentEntry.lastBatchSequence();
+            boolean isAppendAsLeader,
+            SequenceValidation sequenceValidation) {
+        int currentLastSeq = currentLastBatchSequence();
         // must be in sequence, even for the first batch should start from 0
         if (!inSequence(currentLastSeq, appendFirstSeq, isWriterInBatchExpired, isAppendAsLeader)) {
-            throw new OutOfOrderSequenceException(
+            String message =
                     String.format(
                             "Out of order batch sequence for writer %s at offset %s in "
                                     + "table-bucket %s : %s (incoming batch seq.), %s (current batch seq.)",
-                            writerId, lastOffset, tableBucket, appendFirstSeq, currentLastSeq));
+                            writerId, lastOffset, tableBucket, appendFirstSeq, currentLastSeq);
+            if (sequenceValidation == SequenceValidation.WARN_AND_ACCEPT) {
+                LOG.warn("{}. Accepting the persisted batch.", message);
+                return;
+            }
+            throw new OutOfOrderSequenceException(message);
         }
     }
 
+    private int currentLastBatchSequence() {
+        return !updatedEntry.isEmpty()
+                ? updatedEntry.lastBatchSequence()
+                : currentEntry.lastBatchSequence();
+    }
+
     public WriterStateEntry toEntry() {
         return updatedEntry;
     }

fluss-server/src/test/java/org/apache/fluss/server/log/LocalLogTest.java

@@ -32,6 +32,7 @@
 import org.apache.fluss.server.log.LocalLog.SegmentDeletionReason;
 import org.apache.fluss.server.metrics.group.TestingMetricGroups;
 import org.apache.fluss.utils.CloseableIterator;
+import org.apache.fluss.utils.FlussPaths;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -291,6 +292,7 @@ void testCreateAndDeleteSegment() throws Exception {
         assertThat(localLog.getSegments().activeSegment()).isEqualTo(newActiveSegment);
         assertThat(localLog.getSegments().activeSegment()).isNotEqualTo(oldActiveSegment);
         assertThat(localLog.getSegments().activeSegment().getBaseOffset()).isEqualTo(newOffset);
+        assertThat(oldActiveSegment.deleted()).isTrue();
         assertThat(localLog.getRecoveryPoint()).isEqualTo(0L);
         assertThat(localLog.getLocalLogEndOffset()).isEqualTo(newOffset);
         FetchDataInfo read =
@@ -301,6 +303,39 @@ void testCreateAndDeleteSegment() throws Exception {
         assertThat(read.getRecords().sizeInBytes()).isEqualTo(0);
     }
 
+    @Test
+    void testCreateAndDeleteSegmentWithSameOffset() throws Exception {
+        LogSegment oldActiveSegment = localLog.getSegments().activeSegment();
+        oldActiveSegment.offsetIndex();
+        oldActiveSegment.timeIndex();
+        long baseOffset = oldActiveSegment.getBaseOffset();
+        File oldLogFile = oldActiveSegment.getFileLogRecords().file();
+        File oldOffsetIndexFile = oldActiveSegment.getLazyOffsetIndex().file();
+        File oldTimeIndexFile = oldActiveSegment.timeIndexFile();
+        assertThat(oldLogFile).exists();
+        assertThat(oldOffsetIndexFile).exists();
+        assertThat(oldTimeIndexFile).exists();
+
+        LogSegment newActiveSegment =
+                localLog.createAndDeleteSegment(
+                        baseOffset, oldActiveSegment, SegmentDeletionReason.LOG_ROLL);
+
+        assertThat(localLog.getSegments().activeSegment()).isEqualTo(newActiveSegment);
+        assertThat(newActiveSegment.getFileLogRecords().file()).isEqualTo(oldLogFile);
+        assertThat(newActiveSegment.getLazyOffsetIndex().file()).isEqualTo(oldOffsetIndexFile);
+        assertThat(newActiveSegment.timeIndexFile()).isEqualTo(oldTimeIndexFile);
+        assertThat(oldActiveSegment.getFileLogRecords().file().getName())
+                .endsWith(FlussPaths.DELETED_FILE_SUFFIX);
+        assertThat(oldActiveSegment.getLazyOffsetIndex().file().getName())
+                .endsWith(FlussPaths.DELETED_FILE_SUFFIX);
+        assertThat(oldActiveSegment.timeIndexFile().getName())
+                .endsWith(FlussPaths.DELETED_FILE_SUFFIX);
+        assertThat(oldActiveSegment.deleted()).isTrue();
+        assertThat(newActiveSegment.getFileLogRecords().file()).exists();
+        assertThat(newActiveSegment.offsetIndex().file()).exists();
+        assertThat(newActiveSegment.timeIndex().file()).exists();
+    }
+
     @Test
     void testTruncateFullyAndStartAt() throws Exception {
         for (int i = 0; i <= 7; i++) {
@@ -337,6 +372,19 @@ void testTruncateFullyAndStartAt() throws Exception {
         assertThat(read.getRecords().sizeInBytes()).isEqualTo(0);
     }
 
+    @Test
+    void testTruncateFullyAndStartAtDeletesOldActiveSegmentFile() throws Exception {
+        LogSegment oldActiveSegment = localLog.getSegments().activeSegment();
+        File oldLogFile = oldActiveSegment.getFileLogRecords().file();
+        assertThat(oldLogFile).exists();
+
+        localLog.truncateFullyAndStartAt(10L);
+
+        assertThat(localLog.getSegments().baseOffsets()).containsExactly(10L);
+        assertThat(oldActiveSegment.deleted()).isTrue();
+        assertThat(oldLogFile).doesNotExist();
+    }
+
     @Test
     void testTruncateTo() throws Exception {
         for (int i = 0; i <= 11; i++) {

fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java

@@ -319,6 +319,37 @@ void testWriterSnapshotRecoveryFromDiscontinuousBatchSequence() throws Exception
                 .isEqualTo(13);
     }
 
+    @Test
+    void testWriterStateRecoveryAcceptsBatchSequenceGap() throws Exception {
+        LogTablet log = createLogTablet(true);
+        long writerId = 1L;
+
+        log.appendAsFollower(
+                genMemoryLogRecordsWithWriterId(
+                        Collections.singletonList(new Object[] {1, "a"}), writerId, 10, 0L));
+        log.appendAsFollower(
+                genMemoryLogRecordsWithWriterId(
+                        Collections.singletonList(new Object[] {2, "b"}), writerId, 11, 1L));
+        log.roll(Optional.empty());
+
+        MemoryLogRecords recordsWithSequenceGap =
+                genMemoryLogRecordsWithWriterId(
+                        Collections.singletonList(new Object[] {3, "c"}), writerId, 100, 2L);
+        log.activeLogSegment().append(2L, clock.milliseconds(), 2L, recordsWithSequenceGap);
+        log.close();
+
+        log = createLogTablet(false);
+        assertThat(log.localLogEndOffset()).isEqualTo(3L);
+        assertThat(log.writerStateManager().activeWriters().get(writerId).lastBatchSequence())
+                .isEqualTo(100);
+
+        // The recovered state should be persisted in the new snapshot and survive another restart.
+        log.close();
+        log = createLogTablet(false);
+        assertThat(log.writerStateManager().activeWriters().get(writerId).lastBatchSequence())
+                .isEqualTo(100);
+    }
+
     @Test
     void testWriterSnapshotsRecoveryAfterCleanShutdown() throws Exception {
         LogTablet log = createLogTablet(true);

fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java

@@ -366,6 +366,46 @@ void testWriterStateTruncateFullyAndStartAt() throws Exception {
         assertThat(latestWriterSnapshotOffset(log).get()).isEqualTo(29);
     }
 
+    @Test
+    void testTruncateToBeforeFirstSegmentDeletesHigherOffsetSegment() throws Exception {
+        logTablet.truncateFullyAndStartAt(10L);
+        logTablet.appendAsLeader(
+                genMemoryLogRecordsByObject(Collections.singletonList(new Object[] {1, "a"})));
+        LogSegment oldActiveSegment = logTablet.activeLogSegment();
+        assertThat(oldActiveSegment.getBaseOffset()).isEqualTo(10L);
+
+        logTablet.truncateTo(5L);
+
+        assertThat(oldActiveSegment.deleted()).isTrue();
+        assertThat(logTablet.logSegments())
+                .extracting(LogSegment::getBaseOffset)
+                .containsExactly(5L);
+        assertThat(logTablet.localLogEndOffset()).isEqualTo(5L);
+
+        logTablet.close();
+        logTablet =
+                LogTablet.create(
+                        tempDir,
+                        PhysicalTablePath.of(DATA1_TABLE_PATH),
+                        logDir,
+                        conf,
+                        new AtomicBoolean(
+                                conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
+                        TestingMetricGroups.TABLET_SERVER_METRICS,
+                        0,
+                        scheduler,
+                        LogFormat.ARROW,
+                        1,
+                        false,
+                        SystemClock.getInstance(),
+                        false);
+
+        assertThat(logTablet.logSegments())
+                .extracting(LogSegment::getBaseOffset)
+                .containsExactly(5L);
+        assertThat(logTablet.localLogEndOffset()).isEqualTo(5L);
+    }
+
     @Test
     void testWriterIdExpirationOnSegmentDeletion() throws Exception {
         long writerId1 = 1L;