[server] local segment ttl

fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java

@@ -146,6 +146,7 @@ protected void setup() throws Exception {
         guestPrincipal = new FlussPrincipal("guest", "User");
 
         // prepare default database and table
+        FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(3);
         rootAdmin
                 .createDatabase(
                         DATA1_TABLE_PATH_PK.getDatabaseName(), DatabaseDescriptor.EMPTY, true)

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java

@@ -312,7 +312,8 @@ class BlockingFileDownloader extends RemoteFileDownloader {
             protected long downloadFile(Path targetFilePath, FsPath remoteFilePath)
                     throws IOException {
                 int count = enteredCount.incrementAndGet();
-                if (count > 1) {
+                boolean shouldBlock = count > 1;
+                if (shouldBlock) {
                     // Block the 2nd and 3rd downloads to simulate in-flight state.
                     inFlightStarted.countDown();
                     try {
@@ -321,9 +322,13 @@ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath)
                         throw new IOException("Interrupted while blocking", e);
                     }
                 }
-                long downloadBytes = super.downloadFile(targetFilePath, remoteFilePath);
-                inFlightFinished.countDown();
-                return downloadBytes;
+                try {
+                    return super.downloadFile(targetFilePath, remoteFilePath);
+                } finally {
+                    if (shouldBlock) {
+                        inFlightFinished.countDown();
+                    }
+                }
             }
         }
 
@@ -347,8 +352,10 @@ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath)
             // Wait for the first download to complete (already-downloaded file on disk) and 2
             // in-flight downloads to enter the blocked state.
             // At this point: 1 file downloaded, 2 blocked in-flight, 2 pending.
-            retry(Duration.ofMinutes(1), () -> assertThat(futures.get(0).isDone()).isTrue());
             assertThat(inFlightStarted.await(30, TimeUnit.SECONDS)).isTrue();
+            retry(
+                    Duration.ofMinutes(1),
+                    () -> assertThat(futures.subList(0, 3)).anyMatch(future -> future.isDone()));
             Path localLogDir = downloader.getLocalLogDir();
             assertThat(localLogDir.toFile().exists()).isTrue();
 
@@ -362,7 +369,9 @@ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath)
             // Wait for 2 in-flight downloads finished.
             blockLatch.countDown();
             assertThat(inFlightFinished.await(30, TimeUnit.SECONDS)).isTrue();
-            retry(Duration.ofMinutes(1), () -> assertThat(futures.get(1).isDone()).isTrue());
+            retry(
+                    Duration.ofMinutes(1),
+                    () -> assertThat(futures.subList(0, 3)).allMatch(future -> future.isDone()));
 
             // Verify that ultimately the local directory does not exist.
             retry(Duration.ofMinutes(1), () -> assertThat(localLogDir.toFile().exists()).isFalse());

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

@@ -947,6 +947,15 @@ public class ConfigOptions {
                                     + "flush which acts as the log recovery point. The default "
                                     + "setting is 60 seconds.");
 
+    public static final ConfigOption<Duration> LOG_RETENTION_CHECK_INTERVAL =
+            key("log.retention.check-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(5))
+                    .withDescription(
+                            "The frequency with which the log manager checks whether local log "
+                                    + "segments are eligible for TTL cleanup. The value must be "
+                                    + "greater than 0.");
+
     public static final ConfigOption<Duration> LOG_REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL =
             key("log.replica.high-watermark.checkpoint-interval")
                     .durationType()

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

@@ -220,6 +220,7 @@ protected static void validateServerConfigs(Configuration conf) {
         validMinValue(conf, ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, 1);
         validMinValue(conf, ConfigOptions.SERVER_IO_POOL_SIZE, 1);
         validMinValue(conf, ConfigOptions.BACKGROUND_THREADS, 1);
+        validMinDuration(conf, ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, 1);
 
         if (conf.get(ConfigOptions.LOG_SEGMENT_FILE_SIZE).getBytes() > Integer.MAX_VALUE) {
             throw new IllegalConfigurationException(

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

@@ -191,6 +191,30 @@ void testValidateTabletConfigs() {
                 .hasMessageContaining("it must be greater than or equal 0");
     }
 
+    @Test
+    void testValidateLogRetentionCheckInterval() {
+        assertThat(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL.key())
+                .isEqualTo("log.retention.check-interval");
+
+        Configuration conf = new Configuration();
+        conf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+
+        conf.set(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, Duration.ZERO);
+        assertThatThrownBy(() -> validateCoordinatorConfigs(conf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL.key())
+                .hasMessageContaining("must be greater than or equal 1 ms");
+
+        conf.set(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, Duration.ofMillis(-1));
+        assertThatThrownBy(() -> validateCoordinatorConfigs(conf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL.key())
+                .hasMessageContaining("must be greater than or equal 1 ms");
+
+        conf.set(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, Duration.ofMillis(1));
+        validateCoordinatorConfigs(conf);
+    }
+
     @Test
     void testValidateClientConfigs() {
         // valid defaults should pass

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java

@@ -59,11 +59,14 @@
 abstract class FlinkTableSourceBatchITCase extends FlinkTestBase {
 
     static final String CATALOG_NAME = "testcatalog";
-    static final String DEFAULT_DB = "defaultdb";
     protected StreamTableEnvironment tEnv;
+    private String databaseName;
+    private boolean databaseCreated;
 
     @BeforeEach
     void before() {
+        databaseName = null;
+        databaseCreated = false;
         StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
         // create table environment
         tEnv = StreamTableEnvironment.create(execEnv, EnvironmentSettings.inBatchMode());
@@ -75,15 +78,20 @@ void before() {
         tEnv.executeSql("use catalog " + CATALOG_NAME);
 
         tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 4);
-        // create database
-        tEnv.executeSql("create database " + DEFAULT_DB);
-        tEnv.useDatabase(DEFAULT_DB);
+        databaseName = "defaultdb_" + RandomUtils.nextInt();
+        tEnv.executeSql("create database " + databaseName);
+        databaseCreated = true;
+        tEnv.useDatabase(databaseName);
     }
 
     @AfterEach
     void after() {
+        if (tEnv == null || !databaseCreated) {
+            return;
+        }
         tEnv.useDatabase(BUILTIN_DATABASE);
-        tEnv.executeSql(String.format("drop database %s cascade", DEFAULT_DB));
+        tEnv.executeSql(String.format("drop database %s cascade", databaseName));
+        databaseCreated = false;
     }
 
     @Test
@@ -94,10 +102,10 @@ void testScanSingleRowFilter() throws Exception {
         assertThat(tEnv.explainSql(query))
                 .contains(
                         String.format(
-                                "TableSourceScan(table=[[testcatalog, defaultdb, %s, "
+                                "TableSourceScan(table=[[testcatalog, %s, %s, "
                                         + "filter=[and(=(id, 1), =(name, _UTF-16LE'name1':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\"))]]], "
                                         + "fields=[id, address, name])",
-                                tableName));
+                                databaseName, tableName));
         CloseableIterator<Row> collected = tEnv.executeSql(query).collect();
         List<String> expected = Collections.singletonList("+I[1, address1, name1]");
         assertResultsIgnoreOrder(collected, expected, true);
@@ -111,10 +119,10 @@ void testScanSingleRowFilter2() throws Exception {
         assertThat(tEnv.explainSql(query))
                 .contains(
                         String.format(
-                                "TableSourceScan(table=[[testcatalog, defaultdb, %s, "
+                                "TableSourceScan(table=[[testcatalog, %s, %s, "
                                         + "filter=[and(=(id, 1), =(name, _UTF-16LE'name1':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\"))]]], "
                                         + "fields=[id, address, name])",
-                                tableName));
+                                databaseName, tableName));
         CloseableIterator<Row> collected = tEnv.executeSql(query).collect();
         List<String> expected = Collections.singletonList("+I[1, address1, name1]");
         assertResultsIgnoreOrder(collected, expected, true);
@@ -128,10 +136,10 @@ void testScanSingleRowFilter3() throws Exception {
         assertThat(tEnv.explainSql(query))
                 .contains(
                         String.format(
-                                "TableSourceScan(table=[[testcatalog, defaultdb, %s, "
+                                "TableSourceScan(table=[[testcatalog, %s, %s, "
                                         + "filter=[=(id, 1)], "
                                         + "project=[id, name]]], fields=[id, name])",
-                                tableName));
+                                databaseName, tableName));
         CloseableIterator<Row> collected = tEnv.executeSql(query).collect();
         List<String> expected = Collections.singletonList("+I[1, name1]");
         assertResultsIgnoreOrder(collected, expected, true);
@@ -140,7 +148,7 @@ void testScanSingleRowFilter3() throws Exception {
     @Test
     void testScanSingleRowFilterOnPartitionedTable() throws Exception {
         String tableName = prepareSourceTable(new String[] {"id", "dt"}, "dt");
-        TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+        TablePath tablePath = TablePath.of(databaseName, tableName);
         Map<Long, String> partitionNameById =
                 waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), tablePath);
         Iterator<String> partitionIterator =
@@ -152,10 +160,10 @@ void testScanSingleRowFilterOnPartitionedTable() throws Exception {
         assertThat(tEnv.explainSql(query))
                 .contains(
                         String.format(
-                                "TableSourceScan(table=[[testcatalog, defaultdb, %s, "
+                                "TableSourceScan(table=[[testcatalog, %s, %s, "
                                         + "filter=[and(=(id, 1), =(dt, _UTF-16LE'%s':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\"))]]], "
                                         + "fields=[id, address, name, dt])\n",
-                                tableName, partition1));
+                                databaseName, tableName, partition1));
 
         CloseableIterator<Row> collected = tEnv.executeSql(query).collect();
         List<String> expected =
@@ -194,7 +202,7 @@ void testFilterOnLookupSource() throws Exception {
                                 + "  'table.auto-partition.time-unit' = 'year')",
                         dimTableName));
 
-        TablePath srcTablePath = TablePath.of(DEFAULT_DB, srcTableName);
+        TablePath srcTablePath = TablePath.of(databaseName, srcTableName);
         Map<Long, String> partitionNameById =
                 waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), srcTablePath);
         // just pick first partition to insert data
@@ -212,7 +220,7 @@ void testFilterOnLookupSource() throws Exception {
             upsertWriter.flush();
         }
 
-        TablePath dimTablePath = TablePath.of(DEFAULT_DB, dimTableName);
+        TablePath dimTablePath = TablePath.of(databaseName, dimTableName);
         // prepare dim table data
         try (Table dimTable = conn.getTable(dimTablePath)) {
             UpsertWriter upsertWriter = dimTable.newUpsert().createWriter();
@@ -263,7 +271,7 @@ void testLakeTableQueryOnLakeDisabledTable() throws Exception {
                 .isInstanceOf(UnsupportedOperationException.class)
                 .hasMessage(
                         String.format(
-                                "Table %s.%s is not datalake enabled.", DEFAULT_DB, tableName));
+                                "Table %s.%s is not datalake enabled.", databaseName, tableName));
     }
 
     @Test
@@ -461,7 +469,9 @@ void testCountPushDownWithWALMode() throws Exception {
         assertThatThrownBy(() -> tEnv.executeSql(query))
                 .hasRootCauseInstanceOf(InvalidTableException.class)
                 .hasMessageContaining(
-                        "Row count is disabled for this table 'defaultdb.test_count_table_with_wal'.");
+                        String.format(
+                                "Row count is disabled for this table '%s.test_count_table_with_wal'.",
+                                databaseName));
     }
 
     @ParameterizedTest
@@ -546,7 +556,7 @@ private String prepareSourceTable(String[] keys, String partitionedKey) throws E
                             tableName, String.join(",", keys), partitionedKey));
         }
 
-        TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+        TablePath tablePath = TablePath.of(databaseName, tableName);
         String partition1 = null;
         if (partitionedKey != null) {
             Map<Long, String> partitionNameById =
@@ -587,7 +597,7 @@ private String prepareLogTable() throws Exception {
                                 + ")",
                         tableName));
 
-        TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+        TablePath tablePath = TablePath.of(databaseName, tableName);
 
         // prepare table data with NULL values in address column
         try (Table table = conn.getTable(tablePath)) {
@@ -619,7 +629,7 @@ protected String preparePartitionedLogTable() throws Exception {
                                 + "  'table.auto-partition.time-unit' = 'year')",
                         tableName));
 
-        TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+        TablePath tablePath = TablePath.of(databaseName, tableName);
         Map<Long, String> partitionNameById =
                 waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), tablePath);
         Collection<String> partitions = partitionNameById.values();
@@ -657,7 +667,7 @@ private String prepareLogTableWithComplexTypes() throws Exception {
                                 + ")",
                         tableName));
 
-        TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+        TablePath tablePath = TablePath.of(databaseName, tableName);
 
         // prepare table data with complex types
         try (Table table = conn.getTable(tablePath)) {

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

@@ -683,6 +683,7 @@ public interface SegmentDeletionReason {
         SegmentDeletionReason LOG_ROLL = new LogRoll();
         SegmentDeletionReason LOG_DELETION = new LogDeletion();
         SegmentDeletionReason LOG_MOVE_TO_REMOTE = new LogMoveToRemote();
+        SegmentDeletionReason LOG_RETENTION = new LogRetention();
 
         void logReason(List<LogSegment> toDelete);
     }
@@ -718,4 +719,12 @@ public void logReason(List<LogSegment> toDelete) {
             LOG.info("Deleting segments as the log has been moved to remote: " + toDelete);
         }
     }
+
+    /** Delete due to log retention. */
+    private static class LogRetention implements SegmentDeletionReason {
+        @Override
+        public void logReason(List<LogSegment> toDelete) {
+            LOG.info("Deleting segments due to log retention: " + toDelete);
+        }
+    }
 }