[server] Support altering table.log.ttl table option

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

@@ -429,6 +429,49 @@ void testAlterTableConfig() throws Exception {
                 .get();
     }
 
+    @Test
+    void testAlterTableLogTtl() throws Exception {
+        // Verify that altering 'table.log.ttl' is supported: the new value should be persisted in
+        // TableInfo so that subsequent reads observe the updated retention.
+        TablePath tablePath = TablePath.of("test_db", "alter_table_log_ttl");
+        admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, false).get();
+
+        // verify initial value matches the property set in DEFAULT_TABLE_DESCRIPTOR (1 day)
+        TableInfo tableInfo = admin.getTableInfo(tablePath).get();
+        assertThat(tableInfo.getTableConfig().getLogTTLMs())
+                .isEqualTo(Duration.ofDays(1).toMillis());
+
+        // alter to 3d and verify metadata
+        List<TableChange> tableChanges =
+                Collections.singletonList(TableChange.set(ConfigOptions.TABLE_LOG_TTL.key(), "3d"));
+        admin.alterTable(tablePath, tableChanges, false).get();
+
+        tableInfo = admin.getTableInfo(tablePath).get();
+        assertThat(tableInfo.getTableConfig().getLogTTLMs())
+                .isEqualTo(Duration.ofDays(3).toMillis());
+
+        // alter to another value (30d) to verify multiple updates work.
+        tableChanges =
+                Collections.singletonList(
+                        TableChange.set(ConfigOptions.TABLE_LOG_TTL.key(), "30d"));
+        admin.alterTable(tablePath, tableChanges, false).get();
+
+        tableInfo = admin.getTableInfo(tablePath).get();
+        assertThat(tableInfo.getTableConfig().getLogTTLMs())
+                .isEqualTo(Duration.ofDays(30).toMillis());
+
+        // reset to remove the property; value should fall back to the configured default.
+        tableChanges =
+                Collections.singletonList(TableChange.reset(ConfigOptions.TABLE_LOG_TTL.key()));
+        admin.alterTable(tablePath, tableChanges, false).get();
+
+        tableInfo = admin.getTableInfo(tablePath).get();
+        assertThat(tableInfo.toTableDescriptor().getProperties())
+                .doesNotContainKey(ConfigOptions.TABLE_LOG_TTL.key());
+        assertThat(tableInfo.getTableConfig().getLogTTLMs())
+                .isEqualTo(ConfigOptions.TABLE_LOG_TTL.defaultValue().toMillis());
+    }
+
     @Test
     void testAlterTableColumn() throws Exception {
         // create table

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

@@ -51,6 +51,7 @@ public class FlussConfigUtils {
                         ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED.key(),
                         ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(),
                         ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(),
+                        ConfigOptions.TABLE_LOG_TTL.key(),
                         ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS.key(),
                         ConfigOptions.TABLE_AUTO_PARTITION_ENABLED.key(),
                         ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION.key(),

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

@@ -108,7 +108,7 @@ public final class LogTablet {
     private volatile int tieredLogLocalSegments;
     private final Clock clock;
     private final boolean isChangeLog;
-    private final long logTtlMs;
+    private volatile long logTtlMs;
     private final AtomicBoolean rollExpiredActiveSegmentEnabled;
 
     @GuardedBy("lock")
@@ -676,6 +676,11 @@ public void updateTieredLogLocalSegments(int tieredLogLocalSegments) {
         this.tieredLogLocalSegments = tieredLogLocalSegments;
     }
 
+    /** Updates the log ttl; a non-positive value disables local segment expiration. */
+    public void updateLogTtlMs(long logTtlMs) {
+        this.logTtlMs = logTtlMs;
+    }
+
     @VisibleForTesting
     boolean isRollExpiredActiveSegmentEnabled() {
         return rollExpiredActiveSegmentEnabled.get();

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

@@ -290,6 +290,25 @@ public List<RemoteLogSegment> relevantRemoteLogSegmentsForFetchV0(
         return remoteLogTablet(tableBucket).relevantRemoteLogSegmentsForFetchV0(offset);
     }
 
+    /**
+     * Updates the ttl of the {@link RemoteLogTablet} for the given bucket.
+     *
+     * @return the previous ttl, or {@link Optional#empty()} if no tablet is registered (remote
+     *     logging disabled or replica still initializing; the eventually-constructed tablet will
+     *     read the latest ttl from {@code Replica.getLogTTLMs()}).
+     */
+    public Optional<Long> updateLogTtlMs(TableBucket tableBucket, long newTtlMs) {
+        RemoteLogTablet remoteLogTablet = remoteLogs.get(tableBucket);
+        if (remoteLogTablet == null) {
+            return Optional.empty();
+        }
+        long oldTtlMs = remoteLogTablet.getTtlMs();
+        if (oldTtlMs != newTtlMs) {
+            remoteLogTablet.updateTtlMs(newTtlMs);
+        }
+        return Optional.of(oldTtlMs);
+    }
+
     private boolean remoteDisabled() {
         return taskInterval <= 0L;
     }

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

@@ -71,7 +71,7 @@ public class RemoteLogTablet {
     /** The lock to protect the remote log segment list. */
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
 
-    private final long ttlMs;
+    private volatile long ttlMs;
 
     /** The registered metrics for remote log. */
     private volatile MetricGroup remoteLogMetrics;
@@ -161,7 +161,11 @@ public List<RemoteLogSegment> allRemoteLogSegments() {
      */
     public List<RemoteLogSegment> expiredRemoteLogSegments(
             long currentTimeMs, Long lakeLogEndOffset) {
-        if (!logExpireEnable()) {
+        // Snapshot ttlMs to prevent a concurrent update from changing the comparison base
+        // mid-iteration. Without this, an in-flight change to a non-positive value could wrongly
+        // delete all segments.
+        final long ttlSnapshotMs = ttlMs;
+        if (ttlSnapshotMs <= 0) {
             return Collections.emptyList();
         }
         return inReadLock(
@@ -171,7 +175,7 @@ public List<RemoteLogSegment> expiredRemoteLogSegments(
                     for (Map.Entry<Long, Set<UUID>> entry :
                             timestampToRemoteLogSegmentId.entrySet()) {
                         long ts = entry.getKey();
-                        if (currentTimeMs - ts > ttlMs) {
+                        if (currentTimeMs - ts > ttlSnapshotMs) {
                             for (UUID uuid : entry.getValue()) {
                                 RemoteLogSegment segment = idToRemoteLogSegment.get(uuid);
                                 if (lakeLogEndOffset != null) {
@@ -336,8 +340,18 @@ private void addSegment(RemoteLogSegment remoteLogSegment) {
                 .add(remoteLogSegmentId);
     }
 
-    private boolean logExpireEnable() {
-        return ttlMs > 0;
+    /** Returns the current ttl in milliseconds for remote log segments. */
+    public long getTtlMs() {
+        return ttlMs;
+    }
+
+    /**
+     * Update the ttl in milliseconds for remote log segments.
+     *
+     * @param newTtlMs the new ttl in milliseconds; a non-positive value disables expiration
+     */
+    public void updateTtlMs(long newTtlMs) {
+        this.ttlMs = newTtlMs;
     }
 
     private void reset() {

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

@@ -189,6 +189,10 @@ public final class Replica {
     // logFormat and arrowCompressionInfo are used in hot-path, so cache them here.
     private final LogFormat logFormat;
     private final ArrowCompressionInfo arrowCompressionInfo;
+
+    // Caches the latest log TTL so that followers without a RemoteLogTablet retain the value for
+    // leader promotion.
+    private volatile long latestLogTtlMs;
     private final AtomicReference<Integer> leaderReplicaIdOpt = new AtomicReference<>();
     private final ReentrantReadWriteLock leaderIsrUpdateLock = new ReentrantReadWriteLock();
     private final Clock clock;
@@ -274,6 +278,7 @@ public Replica(
         this.tableConfig = tableInfo.getTableConfig();
         this.logFormat = tableConfig.getLogFormat();
         this.arrowCompressionInfo = tableConfig.getArrowCompressionInfo();
+        this.latestLogTtlMs = tableConfig.getLogTTLMs();
         this.snapshotContext = snapshotContext;
         // create a closeable registry for the replica
         this.closeableRegistry = new CloseableRegistry();
@@ -382,7 +387,7 @@ public TableBucket getTableBucket() {
     }
 
     public long getLogTTLMs() {
-        return tableConfig.getLogTTLMs();
+        return latestLogTtlMs;
     }
 
     public int writerIdCount() {
@@ -702,6 +707,37 @@ public void updateTieredLogLocalSegments(int tieredLogLocalSegments) {
                 tieredLogLocalSegments);
     }
 
+    /**
+     * Updates the log ttl when {@code table.log.ttl} is altered. The value is always cached in this
+     * Replica (even for followers without a RemoteLogTablet) so that leader promotion uses the
+     * up-to-date ttl.
+     *
+     * @param newTtlMs the new ttl in milliseconds; a non-positive value disables expiration
+     */
+    public void updateLogTtlMs(long newTtlMs) {
+        long oldValue = latestLogTtlMs;
+        latestLogTtlMs = newTtlMs;
+
+        // Update local LogTablet even when remote logging is disabled.
+        logTablet.updateLogTtlMs(newTtlMs);
+
+        Optional<Long> remoteOldValueOpt = remoteLogManager.updateLogTtlMs(tableBucket, newTtlMs);
+        if (!remoteOldValueOpt.isPresent()) {
+            LOG.debug(
+                    "RemoteLogTablet for {} is unavailable; cached new logTtlMs={} "
+                            + "(remote logging may be disabled or the replica is still initializing).",
+                    tableBucket,
+                    newTtlMs);
+            return;
+        }
+
+        if (oldValue == newTtlMs) {
+            return;
+        }
+
+        LOG.info("Replica for {} logTtlMs changed from {} to {}", tableBucket, oldValue, newTtlMs);
+    }
+
     private void createKv() {
         try {
             // create a closeable registry for the closable related to kv

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

@@ -626,6 +626,7 @@ public void maybeUpdateMetadataCache(int coordinatorEpoch, ClusterMetadata clust
     private void updateReplicaTableConfig(ClusterMetadata clusterMetadata) {
         Map<Long, Boolean> tableIdToLakeFlag = new HashMap<>();
         Map<Long, Integer> tableIdToTieredLogLocalSegments = new HashMap<>();
+        Map<Long, Long> tableIdToLogTtlMs = new HashMap<>();
 
         for (TableMetadata tableMetadata : clusterMetadata.getTableMetadataList()) {
             TableInfo tableInfo = tableMetadata.getTableInfo();
@@ -640,9 +641,15 @@ private void updateReplicaTableConfig(ClusterMetadata clusterMetadata) {
             // Collect tiered log local segments configuration
             int tieredLogLocalSegments = tableInfo.getTableConfig().getTieredLogLocalSegments();
             tableIdToTieredLogLocalSegments.put(tableId, tieredLogLocalSegments);
+
+            // Collect log ttl configuration
+            long logTtlMs = tableInfo.getTableConfig().getLogTTLMs();
+            tableIdToLogTtlMs.put(tableId, logTtlMs);
         }
 
-        if (tableIdToLakeFlag.isEmpty() && tableIdToTieredLogLocalSegments.isEmpty()) {
+        if (tableIdToLakeFlag.isEmpty()
+                && tableIdToTieredLogLocalSegments.isEmpty()
+                && tableIdToLogTtlMs.isEmpty()) {
             return;
         }
 
@@ -662,6 +669,11 @@ private void updateReplicaTableConfig(ClusterMetadata clusterMetadata) {
                     replica.updateTieredLogLocalSegments(
                             tableIdToTieredLogLocalSegments.get(tableId));
                 }
+
+                // Update log ttl configuration
+                if (tableIdToLogTtlMs.containsKey(tableId)) {
+                    replica.updateLogTtlMs(tableIdToLogTtlMs.get(tableId));
+                }
             }
         }
     }

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

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.server.replica.Replica;
 import org.apache.fluss.server.replica.ReplicaTestBase;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -101,4 +102,48 @@ void testExpiredActiveSegmentNotRolledByDefault(boolean partitionTable) throws E
         assertThat(logTablet.localLogStartOffset()).isEqualTo(40L);
         assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L);
     }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testUpdateLogTtlMsAffectsLocalSegmentCleanup(boolean partitionTable) throws Exception {
+        TableBucket tb =
+                partitionTable
+                        ? new TableBucket(DATA1_TABLE_ID, 0L, 0)
+                        : new TableBucket(DATA1_TABLE_ID, 0);
+        conf.set(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED, true);
+        logManager.reconfigure(conf);
+        makeLogTableAsLeader(tb, partitionTable);
+
+        Replica replica = replicaManager.getReplicaOrException(tb);
+        LogTablet logTablet = replica.getLogTablet();
+
+        // Remote log is disabled in this test base, so updateLogTtlMs must still reach LogTablet.
+        assertThatThrownBy(() -> remoteLogManager.remoteLogTablet(tb))
+                .isInstanceOf(IllegalStateException.class);
+
+        addMultiSegmentsToLogTablet(logTablet, 1);
+
+        // Advance 30min — within the 1h TTL, segment is not expired.
+        manualClock.advanceTime(Duration.ofMinutes(30));
+        logManager.cleanupExpiredLocalLogSegments();
+        assertThat(logTablet.getSegments()).hasSize(1);
+
+        // Shrink TTL to 1ms; the segment should now be expired.
+        replica.updateLogTtlMs(1L);
+        logManager.cleanupExpiredLocalLogSegments();
+        // The expired active segment is rolled, creating a new empty segment at offset 10.
+        assertThat(logTablet.getSegments()).hasSize(2);
+        assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(10L);
+
+        // Second pass deletes the now-inactive expired segment.
+        logManager.cleanupExpiredLocalLogSegments();
+        assertThat(logTablet.getSegments()).hasSize(1);
+        assertThat(logTablet.localLogStartOffset()).isEqualTo(10L);
+
+        // Disable expiration; the remaining segment must survive even after a long time.
+        replica.updateLogTtlMs(-1L);
+        manualClock.advanceTime(Duration.ofDays(365));
+        logManager.cleanupExpiredLocalLogSegments();
+        assertThat(logTablet.getSegments()).hasSize(1);
+    }
 }