[server] feat: support table.log.local-ttl

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -865,6 +865,29 @@ void testCreateTableWithInvalidProperty() {
                 .isInstanceOf(InvalidConfigException.class)
                 .hasMessage("'table.log.tiered.local-segments' must be greater than 0.");
 
+        TableDescriptor invalidLocalTtl =
+                TableDescriptor.builder()
+                        .schema(DEFAULT_SCHEMA)
+                        .comment("test table")
+                        .property(ConfigOptions.TABLE_LOG_TTL.key(), "1h")
+                        .property(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "2h")
+                        .build();
+        assertThatThrownBy(() -> admin.createTable(tablePath, invalidLocalTtl, false).get())
+                .cause()
+                .isInstanceOf(InvalidConfigException.class)
+                .hasMessage("'table.log.local-ttl' must be less than or equal to 'table.log.ttl'.");
+
+        TableDescriptor nonPositiveLocalTtl =
+                TableDescriptor.builder()
+                        .schema(DEFAULT_SCHEMA)
+                        .comment("test table")
+                        .property(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "0ms")
+                        .build();
+        assertThatThrownBy(() -> admin.createTable(tablePath, nonPositiveLocalTtl, false).get())
+                .cause()
+                .isInstanceOf(InvalidConfigException.class)
+                .hasMessage("'table.log.local-ttl' must be greater than 0.");
+
         TableDescriptor t4 =
                 TableDescriptor.builder()
                         .schema(DEFAULT_SCHEMA) // no pk
@@ -928,6 +951,24 @@ void testCreateTableWithInvalidProperty() {
                         "Currently, Primary Key Table supports ARROW or COMPACTED log format when kv format is COMPACTED.");
     }
 
+    @Test
+    void testCreateTableWithLocalTtlAndInfiniteRemoteTtl() throws Exception {
+        TablePath tablePath =
+                TablePath.of(
+                        DEFAULT_TABLE_PATH.getDatabaseName(),
+                        "test_local_ttl_with_infinite_remote_ttl");
+        TableDescriptor tableDescriptor =
+                TableDescriptor.builder()
+                        .schema(DEFAULT_SCHEMA)
+                        .property(ConfigOptions.TABLE_LOG_TTL.key(), "0ms")
+                        .property(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "1h")
+                        .build();
+
+        admin.createTable(tablePath, tableDescriptor, false).get();
+        assertThat(admin.tableExists(tablePath).get()).isTrue();
+        admin.dropTable(tablePath, false).get();
+    }
+
     @Test
     void testCreateTableWithInvalidReplicationFactor() throws Exception {
         TablePath tablePath = TablePath.of(DEFAULT_TABLE_PATH.getDatabaseName(), "t1");

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1792,6 +1792,16 @@ public class ConfigOptions {
                             "The number of log segments to retain in local for each table when log tiered storage is enabled. "
                                     + "It must be greater that 0. The default is 2.");
 
+    public static final ConfigOption<Duration> TABLE_LOG_LOCAL_TTL =
+            key("table.log.local-ttl")
+                    .durationType()
+                    .defaultValue(Duration.ofDays(1))
+                    .withDescription(
+                            "The time to retain local log segments after they have been copied to remote storage. "
+                                    + "The default value is 1 day. When configured, the value must be greater "
+                                    + "than 0 and, if `table.log.ttl` is positive, less than or equal to "
+                                    + "`table.log.ttl`.");
+
     public static final ConfigOption<Boolean> TABLE_DATALAKE_ENABLED =
             key("table.datalake.enabled")
                     .booleanType()

fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java

@@ -95,6 +95,11 @@ public int getTieredLogLocalSegments() {
         return config.get(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS);
     }
 
+    /** Gets the TTL of local segments for tiered log. */
+    public long getLocalLogTTLMs() {
+        return config.get(ConfigOptions.TABLE_LOG_LOCAL_TTL).toMillis();
+    }
+
     /** Whether the data lake is enabled. */
     public boolean isDataLakeEnabled() {
         return config.get(ConfigOptions.TABLE_DATALAKE_ENABLED);

fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java

@@ -21,6 +21,8 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for {@link TableConfig}. */
@@ -44,4 +46,16 @@ void testDeleteBehavior() {
         TableConfig tableConfig3 = new TableConfig(conf);
         assertThat(tableConfig3.getDeleteBehavior()).hasValue(DeleteBehavior.IGNORE);
     }
+
+    @Test
+    void testTieredLogLocalTtl() {
+        Configuration conf = new Configuration();
+        conf.set(ConfigOptions.TABLE_LOG_TTL, Duration.ofDays(3));
+        TableConfig tableConfig = new TableConfig(conf);
+
+        assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofDays(1).toMillis());
+
+        conf.set(ConfigOptions.TABLE_LOG_LOCAL_TTL, Duration.ofHours(6));
+        assertThat(tableConfig.getLocalLogTTLMs()).isEqualTo(Duration.ofHours(6).toMillis());
+    }
 }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java

@@ -372,6 +372,15 @@ void testOptionConversions() {
                                 .withDescription(
                                         ConfigOptions.CLIENT_REQUEST_TIMEOUT.description()));
 
+        flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.TABLE_LOG_LOCAL_TTL);
+        assertThat(flinkOption)
+                .isEqualTo(
+                        org.apache.flink.configuration.ConfigOptions.key(
+                                        ConfigOptions.TABLE_LOG_LOCAL_TTL.key())
+                                .stringType()
+                                .defaultValue("86400000 ms")
+                                .withDescription(ConfigOptions.TABLE_LOG_LOCAL_TTL.description()));
+
         flinkOption =
                 FlinkConversions.toFlinkOption(ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE);
         assertThat(flinkOption)

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

@@ -292,7 +292,8 @@ private void waitForLoadLogsInDir(LogRecoveryTask recoveryTask) throws Throwable
      * @param tableBucket the table bucket
      * @param logFormat the log format
      * @param tieredLogLocalSegments the number of segments to retain in local for tiered log
-     * @param logTtlMs the log TTL in milliseconds from table configuration
+     * @param logTtlMs the table log TTL in milliseconds
+     * @param localLogTtlMs the local log TTL in milliseconds from table configuration
      * @param isChangelog whether the log is a changelog of primary key table
      */
     public LogTablet getOrCreateLog(
@@ -302,6 +303,7 @@ public LogTablet getOrCreateLog(
             LogFormat logFormat,
             int tieredLogLocalSegments,
             long logTtlMs,
+            long localLogTtlMs,
             boolean isChangelog)
             throws Exception {
         return inLock(
@@ -324,7 +326,7 @@ public LogTablet getOrCreateLog(
                                     scheduler,
                                     logFormat,
                                     tieredLogLocalSegments,
-                                    logTtlMs,
+                                    effectiveLocalLogTtlMs(logTtlMs, localLogTtlMs),
                                     isChangelog,
                                     clock,
                                     true);
@@ -356,6 +358,7 @@ public LogTablet getOrCreateLog(
                 logFormat,
                 tieredLogLocalSegments,
                 tableConfig.getLogTTLMs(),
+                tableConfig.getLocalLogTTLMs(),
                 isChangelog);
     }
 
@@ -469,7 +472,9 @@ private LogTablet loadLog(
                         scheduler,
                         tableInfo.getTableConfig().getLogFormat(),
                         tableInfo.getTableConfig().getTieredLogLocalSegments(),
-                        tableInfo.getTableConfig().getLogTTLMs(),
+                        effectiveLocalLogTtlMs(
+                                tableInfo.getTableConfig().getLogTTLMs(),
+                                tableInfo.getTableConfig().getLocalLogTTLMs()),
                         tableInfo.hasPrimaryKey(),
                         clock,
                         isCleanShutdown);
@@ -494,6 +499,12 @@ private LogTablet loadLog(
         return logTablet;
     }
 
+    private long effectiveLocalLogTtlMs(long logTtlMs, long localLogTtlMs) {
+        return conf.get(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION).toMillis() > 0L
+                ? localLogTtlMs
+                : logTtlMs;
+    }
+
     /** Close all the logs. */
     public void shutdown() {
         LOG.info("Shutting down LogManager.");

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

@@ -108,7 +108,8 @@ public final class LogTablet {
     private volatile int tieredLogLocalSegments;
     private final Clock clock;
     private final boolean isChangeLog;
-    private final long logTtlMs;
+    private final long localLogTtlMs;
+    private final boolean remoteLogEnabled;
 
     @GuardedBy("lock")
     private volatile LogOffsetMetadata highWatermarkMetadata;
@@ -147,7 +148,7 @@ private LogTablet(
             WriterStateManager writerStateManager,
             LogFormat logFormat,
             int tieredLogLocalSegments,
-            long logTtlMs,
+            long localLogTtlMs,
             boolean isChangelog,
             Clock clock) {
         this.dataDir = dataDir;
@@ -159,7 +160,9 @@ private LogTablet(
                 (int) conf.get(ConfigOptions.WRITER_ID_EXPIRATION_CHECK_INTERVAL).toMillis();
         this.writerStateManager = writerStateManager;
         this.highWatermarkMetadata = new LogOffsetMetadata(0L);
-        this.logTtlMs = logTtlMs;
+        this.localLogTtlMs = localLogTtlMs;
+        this.remoteLogEnabled =
+                conf.get(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION).toMillis() > 0L;
 
         this.scheduler = scheduler;
         // scheduler the writer expiration interval check.
@@ -347,7 +350,7 @@ public static LogTablet create(
             Scheduler scheduler,
             LogFormat logFormat,
             int tieredLogLocalSegments,
-            long logTtlMs,
+            long localLogTtlMs,
             boolean isChangelog,
             Clock clock,
             boolean isCleanShutdown)
@@ -396,7 +399,7 @@ public static LogTablet create(
                 writerStateManager,
                 logFormat,
                 tieredLogLocalSegments,
-                logTtlMs,
+                localLogTtlMs,
                 isChangelog,
                 clock);
     }
@@ -427,7 +430,7 @@ public static LogTablet create(
                 scheduler,
                 logFormat,
                 tieredLogLocalSegments,
-                tableConfig.getLogTTLMs(),
+                tableConfig.getLocalLogTTLMs(),
                 isChangelog,
                 clock,
                 isCleanShutdown);
@@ -742,15 +745,15 @@ public void deleteSegmentsAlreadyExistsInRemote() {
                 this::deletableRemoteSegments);
     }
 
-    /** Deletes inactive local segments that have expired according to the table log TTL. */
+    /**
+     * Deletes inactive local segments that have expired according to the local log TTL.
+     *
+     * <p>When remote log tiering is enabled, cleanup is bounded by the highest offset copied to
+     * remote storage. Otherwise, cleanup is bounded by the high watermark.
+     */
     public void deleteExpiredSegments() {
-        // A missing remote end offset can mean either that no segment has been uploaded or that
-        // all remote segments have expired. In both cases, table.log.ttl remains authoritative for
-        // local retention, while the high watermark and minRetainOffset still protect data that
-        // cannot be deleted yet.
-        long cleanupToOffset = remoteLogEndOffset == -1L ? getHighWatermark() : remoteLogEndOffset;
         deleteSegments(
-                cleanupToOffset,
+                remoteLogEnabled ? remoteLogEndOffset : getHighWatermark(),
                 SegmentDeletionReason.LOG_RETENTION,
                 this::deletableExpiredSegments);
     }
@@ -1362,7 +1365,7 @@ private List<LogSegment> deletableExpiredSegments(long endOffset) throws IOExcep
 
         for (int i = 0; i < logSegments.size() - 1; i++) {
             if (logSegments.get(i + 1).getBaseOffset() > endOffset
-                    || !isSegmentExpired(now, logSegments.get(i), logTtlMs)) {
+                    || !isSegmentExpired(now, logSegments.get(i), localLogTtlMs)) {
                 break;
             }
             deletableSegments.add(logSegments.get(i));

fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java

@@ -357,7 +357,10 @@ public boolean tryToCommitRemoteLogManifest(
         // 2. sending the CommitRemoteLogManifestRequest to coordinator server
         // to try to commit this snapshot.
         long newRemoteLogStartOffset = newRemoteLogManifest.getRemoteLogStartOffset();
-        long newRemoteLogEndOffset = newRemoteLogManifest.getRemoteLogEndOffset();
+        long newRemoteLogEndOffset =
+                Math.max(
+                        newRemoteLogManifest.getRemoteLogEndOffset(),
+                        remoteLogTablet.getHighestRemoteLogEndOffset().orElse(-1L));
         long newRemoteLogSize = newRemoteLogManifest.getRemoteLogSize();
         int retrySendCommitTimes = 1;
         while (retrySendCommitTimes <= 10) {
@@ -436,7 +439,7 @@ private void maybeUpdateCopiedOffset(LogTablet logTablet) {
     }
 
     private long findRemoteLogEndOffset(LogTablet logTablet) {
-        OptionalLong remoteLogEndOffsetOpt = remoteLog.getRemoteLogEndOffset();
+        OptionalLong remoteLogEndOffsetOpt = remoteLog.getHighestRemoteLogEndOffset();
         long newRemoteLogEndOffset;
         if (remoteLogEndOffsetOpt.isPresent()) {
             long remoteLogEndOffset = remoteLogEndOffsetOpt.getAsLong();

fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java

@@ -157,21 +157,50 @@ public void registerReplica(Replica replica) throws Exception {
         Optional<RemoteLogManifestHandle> remoteLogManifestHandleOpt =
                 zkClient.getRemoteLogManifestHandle(tableBucket);
         if (remoteLogManifestHandleOpt.isPresent()) {
+            RemoteLogManifestHandle remoteLogManifestHandle = remoteLogManifestHandleOpt.get();
             // If there is remote log manifest handle in remote, we will download
             // the manifest snapshot from remote storage and write to cache.
             RemoteLogManifest manifest =
                     remoteLogStorage.readRemoteLogManifestSnapshot(
-                            remoteLogManifestHandleOpt.get().getRemoteLogManifestPath());
+                            remoteLogManifestHandle.getRemoteLogManifestPath());
             remoteLog.loadRemoteLogManifest(manifest);
+            remoteLog.updateHighestRemoteLogEndOffset(
+                    remoteLogManifestHandle.getRemoteLogEndOffset());
+            log.updateRemoteLogEndOffset(remoteLogManifestHandle.getRemoteLogEndOffset());
         }
-        remoteLog.getRemoteLogEndOffset().ifPresent(log::updateRemoteLogEndOffset);
         log.updateRemoteLogStartOffset(remoteLog.getRemoteLogStartOffset());
         log.updateRemoteLogSize(remoteLog.getRemoteSizeInBytes());
         // leader needs to register the remote log metrics
         remoteLog.registerMetrics(replica.bucketMetrics());
         remoteLogs.put(tableBucket, remoteLog);
     }
 
+    /**
+     * Best-effort restores the highest copied remote log end offset from persistent metadata.
+     *
+     * <p>A failure only delays local log cleanup until a later remote offset notification and must
+     * not prevent the replica from becoming a follower.
+     */
+    public void restoreRemoteLogEndOffset(Replica replica) {
+        if (remoteDisabled()) {
+            return;
+        }
+        try {
+            Optional<RemoteLogManifestHandle> remoteLogManifestHandle =
+                    zkClient.getRemoteLogManifestHandle(replica.getTableBucket());
+            if (remoteLogManifestHandle.isPresent()) {
+                replica.getLogTablet()
+                        .updateRemoteLogEndOffset(
+                                remoteLogManifestHandle.get().getRemoteLogEndOffset());
+            }
+        } catch (Exception e) {
+            LOG.warn(
+                    "Failed to restore remote log end offset for follower replica {}.",
+                    replica.getTableBucket(),
+                    e);
+        }
+    }
+
     /** Start the log tiering task for the given replica. */
     public void startLogTiering(Replica replica) {
         TableBucket tableBucket = replica.getTableBucket();