diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index 93733b32bb..33c2bc5dc1 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/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 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 diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java index 5f1489baac..c3b04bb165 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java +++ b/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(), diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java index 28dd9a93cc..b82241235d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java +++ b/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(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java index 73bd910df5..2e6e171c93 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java @@ -290,6 +290,25 @@ public List 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 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; } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java index 311931194d..62eb84bc7c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogTablet.java +++ b/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 allRemoteLogSegments() { */ public List 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 expiredRemoteLogSegments( for (Map.Entry> 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() { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java index 784c4ad047..0e0c658237 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java +++ b/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 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 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 diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index 75b0f2ed1c..eda4a68bde 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/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 tableIdToLakeFlag = new HashMap<>(); Map tableIdToTieredLogLocalSegments = new HashMap<>(); + Map 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)); + } } } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java index 237a3a3418..e1923161d0 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java +++ b/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); + } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java index 872682ad26..fe747309ab 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogITCase.java @@ -382,6 +382,103 @@ void testRemoteLogTTLWithDynamicLakeToggle() throws Exception { }); } + @Test + void testAlterTableLogTtlEndToEnd() throws Exception { + TablePath tablePath = TablePath.of("fluss", "test_alter_table_log_ttl_e2e"); + + // Create table with short TTL (1 hour) so we can advance past it quickly. + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(DATA1_SCHEMA) + .distributedBy(1) + .property(ConfigOptions.TABLE_LOG_TTL, Duration.ofHours(1)) + .build(); + + long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor); + TableBucket tb = new TableBucket(tableId, 0); + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb); + + int leaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + TabletServerGateway leaderGateway = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderId); + + // Produce records and wait for remote log copy. + produceRecordsAndWaitRemoteLogCopy(leaderGateway, tb); + + TabletServer leaderServer = FLUSS_CLUSTER_EXTENSION.getTabletServerById(leaderId); + RemoteLogManager remoteLogManager = leaderServer.getReplicaManager().getRemoteLogManager(); + RemoteLogTablet remoteLogTablet = remoteLogManager.remoteLogTablet(tb); + assertThat(remoteLogTablet.allRemoteLogSegments().size()).isGreaterThan(0); + + // --- Real ALTER path: change TTL to 30 days --- + long newTtlMs = Duration.ofDays(30).toMillis(); + CoordinatorGateway coordinatorGateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient(); + coordinatorGateway + .alterTable( + newAlterTableRequest( + tablePath, + Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "30d"), + Collections.emptyList(), + Collections.emptyList(), + false)) + .get(); + + // Wait for metadata propagation: leader's RemoteLogTablet must reflect the new TTL. + retry( + Duration.ofMinutes(1), + () -> assertThat(remoteLogTablet.getTtlMs()).isEqualTo(newTtlMs)); + + // Advance time past the original 1h TTL but well within the new 30d TTL. + MANUAL_CLOCK.advanceTime(Duration.ofHours(1).plusMinutes(30)); + + // Remote segments must NOT be deleted because the new TTL (30d) has not expired. + retry( + Duration.ofMinutes(2), + () -> assertThat(remoteLogTablet.allRemoteLogSegments()).isNotEmpty()); + + // Local segments must also be preserved: the local-only cleaner must use the new TTL. + // With the old 1h TTL, inactive expired segments would be deleted, leaving only the active + // segment (count < tieredLogLocalSegments). With the new 30d TTL, they are retained. + Replica leaderReplica = FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tb); + LogTablet logTablet = leaderReplica.getLogTablet(); + retry( + Duration.ofMinutes(2), + () -> + assertThat(logTablet.getSegments().size()) + .isGreaterThanOrEqualTo(logTablet.getTieredLogLocalSegments())); + + // --- Follower caching: verify a follower cached the new TTL --- + int followerId = (leaderId + 1) % 3; + Replica followerReplica = FLUSS_CLUSTER_EXTENSION.waitAndGetFollowerReplica(tb, followerId); + retry( + Duration.ofMinutes(1), + () -> assertThat(followerReplica.getLogTTLMs()).isEqualTo(newTtlMs)); + + // --- Failover: stop leader, wait for new leader --- + FLUSS_CLUSTER_EXTENSION.stopTabletServer(leaderId); + + int newLeaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + assertThat(newLeaderId).isNotEqualTo(leaderId); + + // The new leader was a follower with no RemoteLogTablet; registerReplica must have + // constructed the new tablet using the cached TTL (30d), not the construction-time + // default (1h). + TabletServer newLeaderServer = FLUSS_CLUSTER_EXTENSION.getTabletServerById(newLeaderId); + RemoteLogManager newRemoteLogManager = + newLeaderServer.getReplicaManager().getRemoteLogManager(); + retry( + Duration.ofMinutes(2), + () -> { + RemoteLogTablet newRemoteLogTablet = newRemoteLogManager.remoteLogTablet(tb); + assertThat(newRemoteLogTablet.getTtlMs()).isEqualTo(newTtlMs); + // Remote segments must still be preserved (new TTL is 30d). + assertThat(newRemoteLogTablet.allRemoteLogSegments()).isNotEmpty(); + }); + + // Restart the stopped server so other tests are not affected. + FLUSS_CLUSTER_EXTENSION.startTabletServer(leaderId); + } + private static Configuration initConfig() { Configuration conf = new Configuration(); conf.setInt(ConfigOptions.DEFAULT_BUCKET_NUMBER, 1); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java index 96e8ae704f..cce469170c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java @@ -791,6 +791,64 @@ void testAlterTableTieredLogLocalSegments(boolean partitionedTable) throws Excep assertThat(logTablet.getSegments()).hasSize(3); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testUpdateLogTtlMsDelegation(boolean partitionedTable) throws Exception { + long tableId = + registerTableInZkClient( + DATA1_TABLE_PATH, + DATA1_SCHEMA, + 201L, + Collections.emptyList(), + Collections.emptyMap()); + TableBucket tb = makeTableBucket(tableId, partitionedTable); + makeLogTableAsLeader(tb, partitionedTable); + + Replica replica = replicaManager.getReplicaOrException(tb); + RemoteLogTablet remoteLog = remoteLogManager.remoteLogTablet(tb); + + // Verify initial ttl matches the configured default. + long defaultTtlMs = ConfigOptions.TABLE_LOG_TTL.defaultValue().toMillis(); + assertThat(remoteLog.getTtlMs()).isEqualTo(defaultTtlMs); + + // Update ttl and verify RemoteLogTablet reflects the new value. + long newTtlMs = Duration.ofDays(1).toMillis(); + replica.updateLogTtlMs(newTtlMs); + assertThat(remoteLog.getTtlMs()).isEqualTo(newTtlMs); + + // no-op: same value must not break anything. + replica.updateLogTtlMs(newTtlMs); + assertThat(remoteLog.getTtlMs()).isEqualTo(newTtlMs); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testLogTtlSurvivesFollowerPromotion(boolean partitionedTable) throws Exception { + long tableId = + registerTableInZkClient( + DATA1_TABLE_PATH, + DATA1_SCHEMA, + 202L, + Collections.emptyList(), + Collections.emptyMap()); + TableBucket tb = makeTableBucket(tableId, partitionedTable); + makeLogTableAsLeader(tb, partitionedTable); + + Replica replica = replicaManager.getReplicaOrException(tb); + + // Simulate follower: no RemoteLogTablet registered. + remoteLogManager.stopLogTiering(replica); + + // TTL update must be cached even without a RemoteLogTablet. + long newTtlMs = Duration.ofDays(30).toMillis(); + replica.updateLogTtlMs(newTtlMs); + assertThat(replica.getLogTTLMs()).isEqualTo(newTtlMs); + + // Simulate promotion: registerReplica must use the cached TTL. + remoteLogManager.registerReplica(replica); + assertThat(remoteLogManager.remoteLogTablet(tb).getTtlMs()).isEqualTo(newTtlMs); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) void testCopySegmentPartialFailureCommitsSuccessfulOnes(boolean partitionTable) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java index 286d1c1262..1f9d7ea58f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java @@ -17,6 +17,7 @@ package org.apache.fluss.server.log.remote; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.remote.RemoteLogManifest; import org.apache.fluss.remote.RemoteLogSegment; import org.apache.fluss.server.log.LogTablet; @@ -26,6 +27,7 @@ import org.junit.jupiter.params.provider.ValueSource; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.UUID; @@ -148,6 +150,36 @@ private void loadRemoteLogSegments( remoteLogSegments)); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testUpdateTtlMs(boolean partitionTable) throws Exception { + LogTablet logTablet = makeLogTabletAndAddSegments(partitionTable); + RemoteLogTablet remoteLogTablet = buildRemoteLogTablet(logTablet); + + // initial ttl follows the default in ConfigOptions.TABLE_LOG_TTL (7 days) + long defaultTtlMs = conf.get(ConfigOptions.TABLE_LOG_TTL).toMillis(); + assertThat(remoteLogTablet.getTtlMs()).isEqualTo(defaultTtlMs); + + // add 1 segment with maxTimestamp = 0 + RemoteLogSegment segment = createLogSegmentWithMaxTimestamp(logTablet, 0L, 0L, 10L); + loadRemoteLogSegments(remoteLogTablet, logTablet, Collections.singletonList(segment)); + + // currentTime = 1 hour. (1h - 0) < 7d, so the segment is NOT expired. + long oneHourMs = java.time.Duration.ofHours(1).toMillis(); + assertThat(remoteLogTablet.expiredRemoteLogSegments(oneHourMs, null)).isEmpty(); + + // shrink ttl to 1 ms via updateTtlMs, the same segment should now be expired. + remoteLogTablet.updateTtlMs(1L); + assertThat(remoteLogTablet.getTtlMs()).isEqualTo(1L); + assertThat(remoteLogTablet.expiredRemoteLogSegments(oneHourMs, null)) + .containsExactly(segment); + + // disable expiration by setting ttl to a non-positive value. + remoteLogTablet.updateTtlMs(-1L); + assertThat(remoteLogTablet.getTtlMs()).isEqualTo(-1L); + assertThat(remoteLogTablet.expiredRemoteLogSegments(oneHourMs, null)).isEmpty(); + } + RemoteLogSegment createLogSegmentWithMaxTimestamp( LogTablet logTablet, long timestamp,