Address feedback

Author: Kaixuan-Duan

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

@@ -453,7 +453,7 @@ void testAlterTableLogTtl() throws Exception {
         assertThat(tableInfo.getTableConfig().getLogTTLMs())
                 .isEqualTo(Duration.ofDays(30).toMillis());
 
-        // reset to remove the property; value should fall back to the default (7 days).
+        // 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();
@@ -462,7 +462,7 @@ void testAlterTableLogTtl() throws Exception {
         assertThat(tableInfo.toTableDescriptor().getProperties())
                 .doesNotContainKey(ConfigOptions.TABLE_LOG_TTL.key());
         assertThat(tableInfo.getTableConfig().getLogTTLMs())
-                .isEqualTo(Duration.ofDays(7).toMillis());
+                .isEqualTo(ConfigOptions.TABLE_LOG_TTL.defaultValue().toMillis());
     }
 
     @Test

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

@@ -275,6 +275,33 @@ public List<RemoteLogSegment> relevantRemoteLogSegments(TableBucket tableBucket,
         return remoteLogTablet(tableBucket).relevantRemoteLogSegments(offset);
     }
 
+    /**
+     * Updates the ttl of the {@link RemoteLogTablet} for the given bucket and returns its previous
+     * ttl, or {@link Optional#empty()} if no tablet is currently registered (which happens in two
+     * normal scenarios):
+     *
+     * <ul>
+     *   <li>remote logging is globally disabled, so no tablet is ever registered;
+     *   <li>the replica is still being created. The eventually-constructed tablet will read its
+     *       initial ttl from {@code Replica.getLogTTLMs()}; whether that reflects the latest ALTER
+     *       depends on the order in which the metadata update arrives at the owning {@code
+     *       ReplicaManager}.
+     * </ul>
+     *
+     * @return the previous ttl in milliseconds, or {@link Optional#empty()} if absent.
+     */
+    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

@@ -169,7 +169,12 @@ public List<RemoteLogSegment> allRemoteLogSegments() {
      */
     public List<RemoteLogSegment> expiredRemoteLogSegments(
             long currentTimeMs, Long lakeLogEndOffset) {
-        if (!logExpireEnable()) {
+        // Snapshot the volatile ttlMs once so that a concurrent updateTtlMs() cannot change the
+        // comparison base mid-iteration. Otherwise an in-flight change to a non-positive value
+        // (which disables expiration) could make currentTimeMs - ts > ttlMs hold for ALL segments
+        // and result in wrongly deleting every remote segment.
+        final long ttlSnapshotMs = ttlMs;
+        if (ttlSnapshotMs <= 0) {
             return Collections.emptyList();
         }
         return inReadLock(
@@ -179,7 +184,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) {
@@ -361,20 +366,13 @@ public void addAndDeleteLogSegments(
                 });
     }
 
-    private boolean logExpireEnable() {
-        return ttlMs > 0;
-    }
-
     /** Returns the current ttl in milliseconds for remote log segments. */
-    @VisibleForTesting
     public long getTtlMs() {
         return ttlMs;
     }
 
     /**
-     * Update the ttl in milliseconds for remote log segments. This is invoked when the user alters
-     * the table option {@code table.log.ttl}. The new value takes effect on the next round of
-     * expired-segment evaluation.
+     * Update the ttl in milliseconds for remote log segments.
      *
      * @param newTtlMs the new ttl in milliseconds; a non-positive value disables expiration
      */

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

@@ -88,7 +88,6 @@
 import org.apache.fluss.server.log.LogTablet;
 import org.apache.fluss.server.log.checkpoint.OffsetCheckpointFile;
 import org.apache.fluss.server.log.remote.RemoteLogManager;
-import org.apache.fluss.server.log.remote.RemoteLogTablet;
 import org.apache.fluss.server.metadata.ServerMetadataCache;
 import org.apache.fluss.server.metadata.TabletServerMetadataCache;
 import org.apache.fluss.server.metrics.group.BucketMetricGroup;
@@ -694,27 +693,21 @@ public void updateTieredLogLocalSegments(int tieredLogLocalSegments) {
      * @param newTtlMs the new ttl in milliseconds; a non-positive value disables expiration
      */
     public void updateLogTtlMs(long newTtlMs) {
-        final RemoteLogTablet remoteLogTablet;
-        try {
-            remoteLogTablet = remoteLogManager.remoteLogTablet(tableBucket);
-        } catch (IllegalStateException e) {
-            // RemoteLogTablet may not be registered yet during the early stage of replica
-            // creation. In that case, the alter is a no-op here because the Replica will read the
-            // up-to-date ttl from the persisted TableInfo when constructing the RemoteLogTablet.
-            LOG.warn(
-                    "RemoteLogTablet for {} is not registered yet, skip applying new logTtlMs={}.",
+        Optional<Long> oldValueOpt = remoteLogManager.updateLogTtlMs(tableBucket, newTtlMs);
+        if (!oldValueOpt.isPresent()) {
+            LOG.debug(
+                    "RemoteLogTablet for {} is unavailable; skip applying new logTtlMs={} "
+                            + "(remote logging may be disabled or the replica is still initializing).",
                     tableBucket,
                     newTtlMs);
             return;
         }
 
-        long oldValue = remoteLogTablet.getTtlMs();
+        long oldValue = oldValueOpt.get();
         if (oldValue == newTtlMs) {
             return;
         }
 
-        remoteLogTablet.updateTtlMs(newTtlMs);
-
         LOG.info("Replica for {} logTtlMs changed from {} to {}", tableBucket, oldValue, newTtlMs);
     }
 

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

@@ -709,8 +709,8 @@ void testAlterTableLogTtl(boolean partitionedTable) throws Exception {
         Replica replica = replicaManager.getReplicaOrException(tb);
         RemoteLogTablet remoteLog = remoteLogManager.remoteLogTablet(tb);
 
-        // Verify initial ttl is 7 days
-        long defaultTtlMs = java.time.Duration.ofDays(7).toMillis();
+        // Verify initial ttl matches the configured default table.log.ttl
+        long defaultTtlMs = ConfigOptions.TABLE_LOG_TTL.defaultValue().toMillis();
         assertThat(remoteLog.getTtlMs()).isEqualTo(defaultTtlMs);
 
         // 2. Update ttl via Replica.updateLogTtlMs (simulating metadata propagation)