[server] Roll expired active segments for remote retention

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

@@ -1048,7 +1048,7 @@ private void maybeRoll(int messageSize, LogAppendInfo appendInfo) throws Excepti
      */
     @VisibleForTesting
     @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
-    public void roll(Optional<Long> expectedNextOffset) throws Exception {
+    public void roll(Optional<Long> expectedNextOffset) throws IOException {
         synchronized (lock) {
             LogSegment segment = localLog.roll(expectedNextOffset);
             // Take a snapshot of the writer state to facilitate recovery. It is useful to have
@@ -1350,7 +1350,12 @@ private List<LogSegment> deletableRemoteSegments(long endOffset) {
         return deletableSegments;
     }
 
-    /** Returns the contiguous prefix of inactive segments that has expired. */
+    /**
+     * Returns the contiguous prefix of expired inactive segments.
+     *
+     * <p>If all inactive segments are deletable, this also checks whether the active segment should
+     * be rolled.
+     */
     private List<LogSegment> deletableExpiredSegments(long endOffset) throws IOException {
         if (localLog.getSegments().isEmpty()) {
             return Collections.emptyList();
@@ -1367,9 +1372,23 @@ private List<LogSegment> deletableExpiredSegments(long endOffset) throws IOExcep
             }
             deletableSegments.add(logSegments.get(i));
         }
+
+        if (deletableSegments.size() == logSegments.size() - 1) {
+            maybeRollExpiredActiveSegment(now, logSegments.get(logSegments.size() - 1));
+        }
         return deletableSegments;
     }
 
+    /** Rolls the active segment when it is non-empty, expired, and fully committed. */
+    private void maybeRollExpiredActiveSegment(long now, LogSegment activeSegment)
+            throws IOException {
+        if (activeSegment.getSizeInBytes() > 0
+                && isSegmentExpired(now, activeSegment, logTtlMs)
+                && getHighWatermark() >= localLogEndOffset()) {
+            roll(Optional.empty());
+        }
+    }
+
     @FunctionalInterface
     private interface DeletableSegmentsFinder {
         List<LogSegment> find(long endOffset) throws IOException;

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

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.log;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.server.replica.ReplicaTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.Collections;
+
+import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
+import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
+import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests TTL-based cleanup when remote log is disabled. */
+final class LocalSegmentTTLTest extends ReplicaTestBase {
+
+    @BeforeEach
+    public void setup() throws Exception {
+        super.setup();
+        registerTableInZkClient(
+                DATA1_TABLE_PATH,
+                DATA1_SCHEMA,
+                DATA1_TABLE_ID,
+                Collections.emptyList(),
+                Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "1h"));
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testExpiredActiveSegmentCleaned(boolean partitionTable) throws Exception {
+        TableBucket tb =
+                partitionTable
+                        ? new TableBucket(DATA1_TABLE_ID, 0L, 0)
+                        : new TableBucket(DATA1_TABLE_ID, 0);
+        makeLogTableAsLeader(tb, partitionTable);
+        LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
+
+        assertThatThrownBy(() -> remoteLogManager.remoteLogTablet(tb))
+                .isInstanceOf(IllegalStateException.class);
+        addMultiSegmentsToLogTablet(logTablet, 1);
+        logManager.cleanupExpiredLocalLogSegments();
+
+        assertThat(logTablet.getSegments()).hasSize(1);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(0L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(0L);
+
+        manualClock.advanceTime(Duration.ofHours(2));
+        logManager.cleanupExpiredLocalLogSegments();
+
+        assertThat(logTablet.getSegments()).hasSize(2);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(0L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(10L);
+
+        logManager.cleanupExpiredLocalLogSegments();
+
+        assertThat(logTablet.getSegments()).hasSize(1);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(10L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(10L);
+    }
+}

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

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.server.log.LogSegment;
 import org.apache.fluss.server.log.LogTablet;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -33,7 +34,7 @@
 import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
 import static org.assertj.core.api.Assertions.assertThat;
 
-/** Tests TTL-based cleanup of inactive local segments retained by tiered storage. */
+/** Tests TTL-based cleanup of local segments retained by tiered storage. */
 final class TieredLocalSegmentTTLTest extends RemoteLogTestBase {
 
     @BeforeEach
@@ -71,17 +72,28 @@ void testInactiveTieredLocalSegmentRemovedAfterTtl(boolean partitionTable) throw
         // Run local retention without updating the remote manifest or uploading another segment.
         logManager.cleanupExpiredLocalLogSegments();
 
-        // The inactive segment is expired and deleted, while the active segment is retained.
+        // The expired active segment is rolled, but is not deleted in the same retention pass.
         assertThat(remoteLog.allRemoteLogSegments()).hasSize(4);
-        assertThat(logTablet.getSegments()).hasSize(1);
+        assertThat(logTablet.getSegments()).hasSize(2);
         assertThat(logTablet.localLogStartOffset()).isEqualTo(40L);
-        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L);
+
+        remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+        assertThat(remoteLog.getRemoteLogEndOffset()).hasValue(50L);
+        logManager.cleanupExpiredLocalLogSegments();
+        assertThat(logTablet.getSegments()).hasSize(1);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(50L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L);
+
+        LogSegment emptyActiveSegment = logTablet.activeLogSegment();
+        logManager.cleanupExpiredLocalLogSegments();
+        assertThat(logTablet.activeLogSegment()).isSameAs(emptyActiveSegment);
     }
 
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
-    void testExpiredLocalSegmentsRemovedWithoutRemoteLogEndOffset(boolean partitionTable)
-            throws Exception {
+    void testExpiredActiveSegmentWaitsForHighWatermarkWithoutRemoteLogEndOffset(
+            boolean partitionTable) throws Exception {
         TableBucket tb =
                 partitionTable
                         ? new TableBucket(DATA1_TABLE_ID, 0L, 0)
@@ -93,11 +105,44 @@ void testExpiredLocalSegmentsRemovedWithoutRemoteLogEndOffset(boolean partitionT
         assertThat(remoteLogManager.remoteLogTablet(tb).getRemoteLogEndOffset()).isEmpty();
 
         manualClock.advanceTime(Duration.ofHours(2));
+        logTablet.updateHighWatermark(logTablet.localLogEndOffset() - 1L);
         logManager.cleanupExpiredLocalLogSegments();
 
         assertThat(logTablet.getSegments()).hasSize(1);
         assertThat(logTablet.localLogStartOffset()).isEqualTo(40L);
         assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L);
+
+        logTablet.updateHighWatermark(logTablet.localLogEndOffset());
+        logManager.cleanupExpiredLocalLogSegments();
+
+        assertThat(logTablet.getSegments()).hasSize(2);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(40L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testExpiredActiveSegmentNotRolledWhenTtlDisabled(boolean partitionTable) throws Exception {
+        registerTableInZkClient(
+                DATA1_TABLE_PATH,
+                DATA1_SCHEMA,
+                DATA1_TABLE_ID,
+                Collections.emptyList(),
+                Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "0ms"));
+        TableBucket tb =
+                partitionTable
+                        ? new TableBucket(DATA1_TABLE_ID, 0L, 0)
+                        : new TableBucket(DATA1_TABLE_ID, 0);
+        makeLogTableAsLeader(tb, partitionTable);
+        LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
+
+        addMultiSegmentsToLogTablet(logTablet, 5);
+        manualClock.advanceTime(Duration.ofHours(2));
+        logManager.cleanupExpiredLocalLogSegments();
+
+        assertThat(logTablet.getSegments()).hasSize(5);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(0L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L);
     }
 
     @ParameterizedTest
@@ -120,5 +165,12 @@ void testTtlCleanupBoundedByRemoteLogEndOffset(boolean partitionTable) throws Ex
         assertThat(logTablet.getSegments()).hasSize(3);
         assertThat(logTablet.localLogStartOffset()).isEqualTo(20L);
         assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L);
+
+        logTablet.updateRemoteLogEndOffset(40L);
+        logManager.cleanupExpiredLocalLogSegments();
+
+        assertThat(logTablet.getSegments()).hasSize(2);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(40L);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L);
     }
 }