You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/08/20 04:01:45 UTC
(fluss) branch main updated (0bd725c05 -> fccf047e9)
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
from 0bd725c05 [rpc] Fix RequestChannel backpressure race
new 21b1bb57d [server] Support altering table.log.ttl table option (#3233)
new e51b5ee93 [test] Verify altered log TTL cleanup end to end
new fccf047e9 [server] Propagate table config updates through TableInfo
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../fluss/client/admin/FlussAdminITCase.java | 39 +++++-
.../table/scanner/log/RemoteLogScannerITCase.java | 75 +++++++++++
.../org/apache/fluss/config/ConfigOptions.java | 4 +-
.../org/apache/fluss/config/FlussConfigUtils.java | 1 +
.../org/apache/fluss/server/log/LogManager.java | 34 ++---
.../org/apache/fluss/server/log/LogTablet.java | 97 +++++++-------
.../fluss/server/log/remote/LogTieringTask.java | 7 +-
.../fluss/server/log/remote/RemoteLogManager.java | 3 +-
.../fluss/server/log/remote/RemoteLogTablet.java | 15 +--
.../org/apache/fluss/server/replica/Replica.java | 115 +++++++----------
.../fluss/server/replica/ReplicaManager.java | 48 ++-----
.../fluss/server/log/LocalSegmentTTLTest.java | 56 +++++++-
.../fluss/server/log/remote/RemoteLogITCase.java | 142 +++++++++++++++++++++
.../server/log/remote/RemoteLogManagerTest.java | 34 ++++-
.../fluss/server/log/remote/RemoteLogTTLTest.java | 7 +-
.../log/remote/RemoteLogTabletOverlapTest.java | 8 +-
.../server/log/remote/RemoteLogTabletTest.java | 28 ++++
.../fluss/server/log/remote/RemoteLogTestBase.java | 5 +-
.../log/remote/TieredLocalSegmentTTLTest.java | 15 ++-
.../replica/NotifyReplicaLakeTableOffsetTest.java | 3 +-
.../apache/fluss/server/replica/ReplicaTest.java | 10 +-
.../fluss/server/replica/ReplicaTestBase.java | 24 ++++
website/docs/engine-flink/options.md | 4 +-
23 files changed, 539 insertions(+), 235 deletions(-)
(fluss) 02/03: [test] Verify altered log TTL cleanup end to end
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit e51b5ee93d60d900265be8ce18af6c203844430c
Author: Jark Wu <ja...@apache.org>
AuthorDate: Wed Aug 19 18:14:41 2026 +0800
[test] Verify altered log TTL cleanup end to end
Cover shortened table log TTL propagation through client fetch, remote
segment expiration, and local cleanup on leader and follower replicas.
---
.../table/scanner/log/RemoteLogScannerITCase.java | 75 ++++++++++++
.../fluss/server/log/remote/RemoteLogITCase.java | 126 ++++++++++++++-------
2 files changed, 157 insertions(+), 44 deletions(-)
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogScannerITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogScannerITCase.java
index 84b50af2c..f31c3cb33 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogScannerITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogScannerITCase.java
@@ -28,10 +28,12 @@ import org.apache.fluss.config.AutoPartitionTimeUnit;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.exception.FetchException;
import org.apache.fluss.metadata.DatabaseDescriptor;
import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.ChangeType;
@@ -39,6 +41,7 @@ import org.apache.fluss.row.GenericRow;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.server.testutils.FlussClusterExtension;
import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.clock.ManualClock;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -49,6 +52,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -58,15 +62,21 @@ import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
import static org.apache.fluss.record.TestData.DATA2_ROW_TYPE;
import static org.apache.fluss.record.TestData.DATA2_TABLE_PATH;
import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** ITCase for {@link LogScannerImpl} as scan log from remote. */
public class RemoteLogScannerITCase {
+ private static final ManualClock MANUAL_CLOCK = new ManualClock(System.currentTimeMillis());
+
@RegisterExtension
public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
FlussClusterExtension.builder()
.setNumOfTabletServers(3)
.setClusterConf(initConfig())
+ .setClock(MANUAL_CLOCK)
.build();
private Connection conn;
@@ -115,6 +125,69 @@ public class RemoteLogScannerITCase {
logScanner.close();
}
+ @Test
+ void testAlterTableLogTtlAffectsFetch() throws Exception {
+ TablePath tablePath = TablePath.of("test_db_1", "test_alter_table_log_ttl_affects_fetch");
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(DATA1_SCHEMA)
+ .distributedBy(1)
+ .property(ConfigOptions.TABLE_LOG_TTL, Duration.ofDays(30))
+ .build();
+ long tableId = createTable(tablePath, tableDescriptor);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+
+ try (Table table = conn.getTable(tablePath)) {
+ AppendWriter appendWriter = table.newAppend().createWriter();
+ int oldRecordSize = 20;
+ List<GenericRow> expectedOldRows = new ArrayList<>();
+ for (int i = 0; i < oldRecordSize; i++) {
+ GenericRow oldRow = row(i, "old_" + i);
+ expectedOldRows.add(oldRow);
+ appendWriter.append(oldRow).get();
+ }
+ FLUSS_CLUSTER_EXTENSION.waitUntilSomeLogSegmentsCopyToRemote(tableBucket);
+
+ // assert all the records can be fetched before altering the TTL.
+ List<GenericRow> actualOldRows = new ArrayList<>();
+ try (LogScanner logScanner = table.newScan().createLogScanner()) {
+ logScanner.subscribeFromBeginning(0);
+ waitUntil(
+ () -> {
+ for (ScanRecord scanRecord : logScanner.poll(Duration.ofSeconds(1))) {
+ InternalRow actualRow = scanRecord.getRow();
+ actualOldRows.add(row(actualRow.getInt(0), actualRow.getString(1)));
+ }
+ return actualOldRows.size() >= oldRecordSize;
+ },
+ Duration.ofMinutes(1),
+ "Timed out waiting for records appended before altering the TTL.");
+ }
+ assertThat(actualOldRows).containsExactlyElementsOf(expectedOldRows);
+
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.set(ConfigOptions.TABLE_LOG_TTL.key(), "1h")),
+ false)
+ .get();
+ MANUAL_CLOCK.advanceTime(Duration.ofHours(1).plusMinutes(30));
+
+ // Retry with a fresh scanner until the last record written before the TTL change is
+ // unavailable from both local and remote logs.
+ retry(
+ Duration.ofMinutes(2),
+ () -> {
+ try (LogScanner expiredLogScanner = table.newScan().createLogScanner()) {
+ expiredLogScanner.subscribe(0, oldRecordSize - 1L);
+ assertThatThrownBy(() -> expiredLogScanner.poll(Duration.ofSeconds(1)))
+ .isInstanceOf(FetchException.class)
+ .hasMessageContaining("is out of range");
+ }
+ });
+ }
+ }
+
@ParameterizedTest
@ValueSource(strings = {"INDEXED", "ARROW"})
void testScanFromRemoteAndProject(String format) throws Exception {
@@ -270,6 +343,8 @@ public class RemoteLogScannerITCase {
conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 3);
// set a shorter interval for testing purpose
conf.set(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION, Duration.ofSeconds(1));
+ conf.set(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, Duration.ofSeconds(1));
+ conf.set(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED, true);
conf.set(ConfigOptions.LOG_SEGMENT_FILE_SIZE, MemorySize.parse("1kb"));
return conf;
}
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 fe747309a..514dd8f77 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
@@ -386,12 +386,13 @@ public class RemoteLogITCase {
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.
+ // Keep enough local segments so that only TTL cleanup can advance local-log start offsets.
TableDescriptor tableDescriptor =
TableDescriptor.builder()
.schema(DATA1_SCHEMA)
.distributedBy(1)
- .property(ConfigOptions.TABLE_LOG_TTL, Duration.ofHours(1))
+ .property(ConfigOptions.TABLE_LOG_TTL, Duration.ofDays(30))
+ .property(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, 100)
.build();
long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor);
@@ -402,81 +403,117 @@ public class RemoteLogITCase {
TabletServerGateway leaderGateway =
FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderId);
- // Produce records and wait for remote log copy.
+ // Produce the old segment cohort and wait for it to be copied to remote storage.
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);
+ List<String> oldRemoteSegmentIds =
+ remoteLogTablet.allRemoteLogSegments().stream()
+ .map(segment -> segment.remoteLogSegmentId().toString())
+ .collect(Collectors.toList());
+ assertThat(oldRemoteSegmentIds).isNotEmpty();
+
+ Replica leaderReplica = FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tb);
+ LogTablet leaderLogTablet = leaderReplica.getLogTablet();
+ long firstBatchEndOffset = leaderLogTablet.localLogEndOffset();
+
+ int followerId = (leaderId + 1) % 3;
+ Replica followerReplica = FLUSS_CLUSTER_EXTENSION.waitAndGetFollowerReplica(tb, followerId);
+ LogTablet followerLogTablet = followerReplica.getLogTablet();
+ retry(
+ Duration.ofMinutes(1),
+ () ->
+ assertThat(followerLogTablet.localLogEndOffset())
+ .isEqualTo(firstBatchEndOffset));
- // --- Real ALTER path: change TTL to 30 days ---
- long newTtlMs = Duration.ofDays(30).toMillis();
+ long initialLeaderLocalStartOffset = leaderLogTablet.localLogStartOffset();
+ long initialFollowerLocalStartOffset = followerLogTablet.localLogStartOffset();
+
+ // Shorten the TTL through the real ALTER path.
+ long newTtlMs = Duration.ofHours(1).toMillis();
CoordinatorGateway coordinatorGateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
coordinatorGateway
.alterTable(
newAlterTableRequest(
tablePath,
- Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "30d"),
+ Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "1h"),
Collections.emptyList(),
Collections.emptyList(),
false))
.get();
- // Wait for metadata propagation: leader's RemoteLogTablet must reflect the new TTL.
+ // Wait until the new TTL reaches the leader's remote and local tablets and the follower's
+ // local tablet.
retry(
Duration.ofMinutes(1),
- () -> assertThat(remoteLogTablet.getTtlMs()).isEqualTo(newTtlMs));
+ () -> {
+ assertThat(remoteLogTablet.getTtlMs()).isEqualTo(newTtlMs);
+ assertThat(leaderLogTablet.getEffectiveLocalLogTtlMs()).isEqualTo(newTtlMs);
+ assertThat(followerReplica.getLogTTLMs()).isEqualTo(newTtlMs);
+ assertThat(followerLogTablet.getEffectiveLocalLogTtlMs()).isEqualTo(newTtlMs);
+ });
- // Advance time past the original 1h TTL but well within the new 30d TTL.
+ // Expire the old cohort, then append a new cohort that must remain available.
MANUAL_CLOCK.advanceTime(Duration.ofHours(1).plusMinutes(30));
+ produceRecordsAndWaitRemoteLogCopy(leaderGateway, tb, firstBatchEndOffset);
- // Remote segments must NOT be deleted because the new TTL (30d) has not expired.
+ // The old remote segments must be removed while newly uploaded segments remain.
retry(
Duration.ofMinutes(2),
- () -> assertThat(remoteLogTablet.allRemoteLogSegments()).isNotEmpty());
+ () -> {
+ List<RemoteLogSegment> remainingSegments =
+ remoteLogTablet.allRemoteLogSegments();
+ assertThat(remainingSegments).isNotEmpty();
+ assertThat(remainingSegments)
+ .extracting(segment -> segment.remoteLogSegmentId().toString())
+ .doesNotContainAnyElementsOf(oldRemoteSegmentIds);
+ assertThat(remainingSegments)
+ .allMatch(
+ segment -> segment.remoteLogEndOffset() > firstBatchEndOffset);
+ });
- // 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();
+ // Wait for the second cohort to reach the follower and for LogManager's retention task to
+ // automatically discard the expired local prefix on both replicas.
retry(
Duration.ofMinutes(2),
- () ->
- assertThat(logTablet.getSegments().size())
- .isGreaterThanOrEqualTo(logTablet.getTieredLogLocalSegments()));
+ () -> {
+ assertThat(followerLogTablet.localLogEndOffset())
+ .isGreaterThan(firstBatchEndOffset);
+ assertThat(leaderLogTablet.localLogStartOffset())
+ .isGreaterThan(initialLeaderLocalStartOffset);
+ assertThat(followerLogTablet.localLogStartOffset())
+ .isGreaterThan(initialFollowerLocalStartOffset);
+ });
- // --- Follower caching: verify a follower cached the new TTL ---
- int followerId = (leaderId + 1) % 3;
- Replica followerReplica = FLUSS_CLUSTER_EXTENSION.waitAndGetFollowerReplica(tb, followerId);
+ // Verify follower caching still preserves the shortened TTL through leader promotion.
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);
+ try {
+ int newLeaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb);
+ assertThat(newLeaderId).isNotEqualTo(leaderId);
+
+ // The promoted follower must construct its RemoteLogTablet with the altered TTL.
+ 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);
+ assertThat(newRemoteLogTablet.allRemoteLogSegments()).isNotEmpty();
+ });
+ } finally {
+ // Restart the stopped server so other tests are not affected.
+ FLUSS_CLUSTER_EXTENSION.startTabletServer(leaderId);
+ }
}
private static Configuration initConfig() {
@@ -485,6 +522,7 @@ public class RemoteLogITCase {
conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 3);
// set a shorter interval for testing purpose
conf.set(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION, Duration.ofSeconds(1));
+ conf.set(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, Duration.ofSeconds(1));
conf.set(ConfigOptions.LOG_SEGMENT_FILE_SIZE, MemorySize.parse("1kb"));
// set a shorter max log time to allow replica shrink from isr. Don't be too low, otherwise
(fluss) 01/03: [server] Support altering table.log.ttl table option (#3233)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 21b1bb57dbe4713e25f38e43ed4cbd74adb4588e
Author: duankaixuan <14...@qq.com>
AuthorDate: Wed Apr 29 17:13:30 2026 +0800
[server] Support altering table.log.ttl table option (#3233)
---
.../fluss/client/admin/FlussAdminITCase.java | 39 ++++++++-
.../org/apache/fluss/config/FlussConfigUtils.java | 1 +
.../fluss/server/log/remote/RemoteLogManager.java | 19 +++++
.../fluss/server/log/remote/RemoteLogTablet.java | 24 ++++--
.../org/apache/fluss/server/replica/Replica.java | 36 ++++++--
.../fluss/server/log/LocalSegmentTTLTest.java | 45 ++++++++++
.../fluss/server/log/remote/RemoteLogITCase.java | 97 ++++++++++++++++++++++
.../server/log/remote/RemoteLogManagerTest.java | 58 +++++++++++++
.../server/log/remote/RemoteLogTabletTest.java | 32 +++++++
9 files changed, 338 insertions(+), 13 deletions(-)
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 943a44a56..7b8cd0058 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,9 +429,46 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.get();
}
+ @Test
+ void testAlterTableLogTtl() throws Exception {
+ TablePath tablePath = TablePath.of("test_db", "alter_table_log_ttl");
+ admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, false).get();
+ // verify initial value matches 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 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; value falls back to 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
TablePath tablePath = TablePath.of("test_db", "alter_table_1");
admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, false).get();
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 f48748ca1..9a7ff3cd5 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_LOG_LOCAL_TTL.key(),
ConfigOptions.TABLE_AUTO_PARTITION_ENABLED.key(),
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 73bd910df..2e6e171c9 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 class RemoteLogManager implements Closeable {
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;
}
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 311931194..62eb84bc7 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 class RemoteLogTablet {
*/
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 class RemoteLogTablet {
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 @@ public class RemoteLogTablet {
.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 8ba20b4d9..47eb0e7b7 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<Integer> leaderReplicaIdOpt = new AtomicReference<>();
private final ReentrantReadWriteLock leaderIsrUpdateLock = new ReentrantReadWriteLock();
private final Clock clock;
@@ -274,6 +278,7 @@ public final class 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 final class Replica {
}
public long getLogTTLMs() {
- return tableConfig.getLogTTLMs();
+ return latestLogTtlMs;
}
public int writerIdCount() {
@@ -702,20 +707,37 @@ public final class Replica {
tieredLogLocalSegments);
}
- /** Updates the log TTLs after the table configuration is altered. */
+ /**
+ * Updates the log TTLs after the table configuration 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.
+ */
public void updateLogTtls(long logTtlMs, long localLogTtlMs) {
- long oldValue = logTablet.getEffectiveLocalLogTtlMs();
+ latestLogTtlMs = logTtlMs;
+
+ long oldEffectiveTtl = logTablet.getEffectiveLocalLogTtlMs();
logTablet.updateLogTtls(logTtlMs, localLogTtlMs);
- long newValue = logTablet.getEffectiveLocalLogTtlMs();
- if (oldValue == newValue) {
+ long newEffectiveTtl = logTablet.getEffectiveLocalLogTtlMs();
+
+ Optional<Long> remoteOldValueOpt = remoteLogManager.updateLogTtlMs(tableBucket, logTtlMs);
+ if (!remoteOldValueOpt.isPresent()) {
+ LOG.debug(
+ "RemoteLogTablet for {} is unavailable; cached new logTtlMs={} "
+ + "(remote logging may be disabled or the replica is still initializing).",
+ tableBucket,
+ logTtlMs);
+ return;
+ }
+
+ if (oldEffectiveTtl == newEffectiveTtl) {
return;
}
LOG.info(
"Replica for {} effectiveLocalLogTtlMs changed from {} to {}",
tableBucket,
- oldValue,
- newValue);
+ oldEffectiveTtl,
+ newEffectiveTtl);
}
private void createKv() {
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 a8a42ffe8..68507a170 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 @@ package org.apache.fluss.server.log;
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;
@@ -128,4 +129,48 @@ final class LocalSegmentTTLTest extends ReplicaTestBase {
assertThat(logTablet.getEffectiveLocalLogTtlMs()).isEqualTo(Duration.ofHours(2).toMillis());
}
+
+ @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 updateLogTtls 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.updateLogTtls(1L, 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.updateLogTtls(-1L, -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 872682ad2..fe747309a 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 @@ public class RemoteLogITCase {
});
}
+ @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 96e8ae704..139e762ca 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 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
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.updateLogTtls(newTtlMs, newTtlMs);
+ assertThat(remoteLog.getTtlMs()).isEqualTo(newTtlMs);
+
+ // no-op: same value must not break anything.
+ replica.updateLogTtls(newTtlMs, 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.updateLogTtls(newTtlMs, 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 286d1c126..1f9d7ea58 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.ParameterizedTest;
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 @@ class RemoteLogTabletTest extends RemoteLogTestBase {
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,
(fluss) 03/03: [server] Propagate table config updates through TableInfo
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit fccf047e974a9feb272cfe64211dc3e4af98981d
Author: Jark Wu <ja...@apache.org>
AuthorDate: Wed Aug 19 20:44:52 2026 +0800
[server] Propagate table config updates through TableInfo
Replace per-option replica updates with a full TableInfo snapshot and
derive mutable log-tablet settings from TableConfig.
AI-Contributed/Feature: 0/19
AI-Contributed/UT: 0/2
---
.../org/apache/fluss/config/ConfigOptions.java | 4 +-
.../org/apache/fluss/server/log/LogManager.java | 34 ++---
.../org/apache/fluss/server/log/LogTablet.java | 97 +++++++--------
.../fluss/server/log/remote/LogTieringTask.java | 7 +-
.../fluss/server/log/remote/RemoteLogManager.java | 22 +---
.../fluss/server/log/remote/RemoteLogTablet.java | 31 +----
.../org/apache/fluss/server/replica/Replica.java | 137 +++++++--------------
.../fluss/server/replica/ReplicaManager.java | 48 ++------
.../fluss/server/log/LocalSegmentTTLTest.java | 17 ++-
.../fluss/server/log/remote/RemoteLogITCase.java | 23 ++--
.../server/log/remote/RemoteLogManagerTest.java | 48 ++------
.../fluss/server/log/remote/RemoteLogTTLTest.java | 7 +-
.../log/remote/RemoteLogTabletOverlapTest.java | 8 +-
.../server/log/remote/RemoteLogTabletTest.java | 20 ++-
.../fluss/server/log/remote/RemoteLogTestBase.java | 5 +-
.../log/remote/TieredLocalSegmentTTLTest.java | 15 ++-
.../replica/NotifyReplicaLakeTableOffsetTest.java | 3 +-
.../apache/fluss/server/replica/ReplicaTest.java | 10 +-
.../fluss/server/replica/ReplicaTestBase.java | 24 ++++
website/docs/engine-flink/options.md | 4 +-
20 files changed, 215 insertions(+), 349 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index e912aa1d4..1a2cbd698 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1845,7 +1845,7 @@ public class ConfigOptions {
+ "maximum time log segments are retained before they become "
+ "eligible for deletion. When remote log tiering is enabled, "
+ "this value controls the retention of remote log segments. "
- + "Setting the value to -1 disables TTL-based deletion. "
+ + "Setting the value to '0ms' disables TTL-based deletion. "
+ "The default value is 7 days.");
public static final ConfigOption<Integer> TABLE_TIERED_LOG_LOCAL_SEGMENTS =
@@ -1865,7 +1865,7 @@ public class ConfigOptions {
+ "maximum time local log segments are retained before they become "
+ "eligible for deletion. When remote log tiering is enabled, an "
+ "expired local segment is deleted only after it has been copied "
- + "to remote storage. Setting the value to -1 disables TTL-based "
+ + "to remote storage. Setting the value to '0ms' disables TTL-based "
+ "deletion. If not configured, the value inherits `table.log.ttl`. "
+ "When both values are positive, it must be less than or equal to "
+ "`table.log.ttl`.");
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java
index 2e714e20c..4c1a84dd1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java
@@ -296,20 +296,14 @@ public final class LogManager extends TabletManagerBase implements ServerReconfi
* @param dataDir the local data directory chosen for the bucket
* @param tablePath the table path of the bucket belongs to
* @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 table log TTL in milliseconds
- * @param localLogTtlMs the local log TTL in milliseconds from table configuration
+ * @param tableConfig the table configuration
* @param isChangelog whether the log is a changelog of primary key table
*/
public LogTablet getOrCreateLog(
File dataDir,
PhysicalTablePath tablePath,
TableBucket tableBucket,
- LogFormat logFormat,
- int tieredLogLocalSegments,
- long logTtlMs,
- long localLogTtlMs,
+ TableConfig tableConfig,
boolean isChangelog)
throws Exception {
return inLock(
@@ -331,10 +325,7 @@ public final class LogManager extends TabletManagerBase implements ServerReconfi
serverMetricGroup,
0L,
scheduler,
- logFormat,
- tieredLogLocalSegments,
- logTtlMs,
- localLogTtlMs,
+ tableConfig,
isChangelog,
clock,
true);
@@ -358,16 +349,11 @@ public final class LogManager extends TabletManagerBase implements ServerReconfi
int tieredLogLocalSegments,
boolean isChangelog)
throws Exception {
- TableConfig tableConfig = new TableConfig(new Configuration());
+ Configuration tableProperties = new Configuration();
+ tableProperties.set(ConfigOptions.TABLE_LOG_FORMAT, logFormat);
+ tableProperties.set(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, tieredLogLocalSegments);
return getOrCreateLog(
- dataDir,
- tablePath,
- tableBucket,
- logFormat,
- tieredLogLocalSegments,
- tableConfig.getLogTTLMs(),
- tableConfig.getLocalLogTTLMs(),
- isChangelog);
+ dataDir, tablePath, tableBucket, new TableConfig(tableProperties), isChangelog);
}
public Optional<LogTablet> getLog(TableBucket tableBucket) {
@@ -479,14 +465,10 @@ public final class LogManager extends TabletManagerBase implements ServerReconfi
serverMetricGroup,
logRecoveryPoint,
scheduler,
- tableInfo.getTableConfig().getLogFormat(),
- tableInfo.getTableConfig().getTieredLogLocalSegments(),
- tableInfo.getTableConfig().getLogTTLMs(),
- tableInfo.getTableConfig().getLocalLogTTLMs(),
+ tableInfo.getTableConfig(),
tableInfo.hasPrimaryKey(),
clock,
isCleanShutdown);
- logTablet.updateIsDataLakeEnabled(tableInfo.getTableConfig().isDataLakeEnabled());
if (currentLogs.containsKey(tableBucket)) {
throw new IllegalStateException(
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 d4f880b89..386e8b416 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
@@ -105,13 +105,15 @@ public final class LogTablet {
private final Scheduler scheduler;
private final ScheduledFuture<?> writerExpireCheck;
private final LogFormat logFormat;
+ // Mutable values derived from the latest table configuration.
private volatile int tieredLogLocalSegments;
+ private volatile long effectiveLocalLogTtlMs;
+ private volatile boolean isDataLakeEnabled;
private final Clock clock;
private final boolean isChangeLog;
private final AtomicBoolean rollExpiredActiveSegmentEnabled;
private final boolean remoteLogEnabled;
- private volatile long effectiveLocalLogTtlMs;
@GuardedBy("lock")
private volatile LogOffsetMetadata highWatermarkMetadata;
@@ -131,8 +133,6 @@ public final class LogTablet {
// tracking the log size in remote storage
private volatile long remoteLogSize = 0;
- // tracking if the data lake enabled
- private volatile boolean isDataLakeEnabled = false;
// tracking the log start/end offset in lakehouse storage
private volatile long lakeTableSnapshotId = -1;
// note: currently, for primary key table, the log start offset nerve be updated
@@ -151,10 +151,7 @@ public final class LogTablet {
AtomicBoolean rollExpiredActiveSegmentEnabled,
Scheduler scheduler,
WriterStateManager writerStateManager,
- LogFormat logFormat,
- int tieredLogLocalSegments,
- long logTtlMs,
- long localLogTtlMs,
+ TableConfig tableConfig,
boolean isChangelog,
Clock clock) {
this.dataDir = dataDir;
@@ -172,7 +169,8 @@ public final class LogTablet {
"rollExpiredActiveSegmentEnabled must not be null");
this.remoteLogEnabled =
conf.get(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION).toMillis() > 0L;
- this.effectiveLocalLogTtlMs = effectiveLocalLogTtlMs(logTtlMs, localLogTtlMs);
+ applyTableConfig(tableConfig);
+ this.logFormat = tableConfig.getLogFormat();
this.scheduler = scheduler;
// scheduler the writer expiration interval check.
@@ -182,12 +180,6 @@ public final class LogTablet {
() -> removeExpiredWriter(System.currentTimeMillis()),
writerExpirationCheckIntervalMs,
writerExpirationCheckIntervalMs);
- this.logFormat = logFormat;
- checkArgument(
- tieredLogLocalSegments > 0,
- "log segments to retain in local must be greater than 0");
- this.tieredLogLocalSegments = tieredLogLocalSegments;
-
this.clock = clock;
this.isChangeLog = isChangelog;
// Default value to 0L for changelog to avoid cleaning up any segments in case of not
@@ -341,6 +333,22 @@ public final class LogTablet {
return logFormat;
}
+ /** Applies the mutable runtime configuration values used by this log tablet. */
+ public void applyTableConfig(TableConfig tableConfig) {
+ TableConfig newTableConfig = checkNotNull(tableConfig, "tableConfig");
+ int tieredLogLocalSegments = newTableConfig.getTieredLogLocalSegments();
+ checkArgument(
+ tieredLogLocalSegments > 0,
+ "log segments to retain in local must be greater than 0");
+ long effectiveLocalLogTtlMs =
+ remoteLogEnabled ? newTableConfig.getLocalLogTTLMs() : newTableConfig.getLogTTLMs();
+ boolean isDataLakeEnabled = newTableConfig.isDataLakeEnabled();
+
+ this.tieredLogLocalSegments = tieredLogLocalSegments;
+ this.effectiveLocalLogTtlMs = effectiveLocalLogTtlMs;
+ this.isDataLakeEnabled = isDataLakeEnabled;
+ }
+
public long getLeaderEndOffsetSnapshot() {
return leaderEndOffsetSnapshot;
}
@@ -359,14 +367,14 @@ public final class LogTablet {
TabletServerMetricGroup serverMetricGroup,
long recoveryPoint,
Scheduler scheduler,
- LogFormat logFormat,
- int tieredLogLocalSegments,
- long logTtlMs,
- long localLogTtlMs,
+ TableConfig tableConfig,
boolean isChangelog,
Clock clock,
boolean isCleanShutdown)
throws Exception {
+ checkNotNull(tableConfig, "tableConfig");
+ LogFormat logFormat = tableConfig.getLogFormat();
+
// create the log directory if it doesn't exist
Files.createDirectories(tabletDir.toPath());
@@ -410,10 +418,7 @@ public final class LogTablet {
rollExpiredActiveSegmentEnabled,
scheduler,
writerStateManager,
- logFormat,
- tieredLogLocalSegments,
- logTtlMs,
- localLogTtlMs,
+ tableConfig,
isChangelog,
clock);
}
@@ -434,7 +439,9 @@ public final class LogTablet {
Clock clock,
boolean isCleanShutdown)
throws Exception {
- TableConfig tableConfig = new TableConfig(new Configuration());
+ Configuration tableProperties = new Configuration();
+ tableProperties.set(ConfigOptions.TABLE_LOG_FORMAT, logFormat);
+ tableProperties.set(ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, tieredLogLocalSegments);
return create(
dataDir,
tablePath,
@@ -444,10 +451,7 @@ public final class LogTablet {
serverMetricGroup,
recoveryPoint,
scheduler,
- logFormat,
- tieredLogLocalSegments,
- tableConfig.getLogTTLMs(),
- tableConfig.getLocalLogTTLMs(),
+ new TableConfig(tableProperties),
isChangelog,
clock,
isCleanShutdown);
@@ -676,14 +680,6 @@ public final class LogTablet {
}
}
- public void updateIsDataLakeEnabled(boolean isDataLakeEnabled) {
- this.isDataLakeEnabled = isDataLakeEnabled;
- }
-
- public void updateTieredLogLocalSegments(int tieredLogLocalSegments) {
- this.tieredLogLocalSegments = tieredLogLocalSegments;
- }
-
@VisibleForTesting
boolean isRollExpiredActiveSegmentEnabled() {
return rollExpiredActiveSegmentEnabled.get();
@@ -693,11 +689,6 @@ public final class LogTablet {
return tieredLogLocalSegments;
}
- /** Updates the remote and local log TTLs used to derive the effective local log TTL. */
- public void updateLogTtls(long logTtlMs, long localLogTtlMs) {
- this.effectiveLocalLogTtlMs = effectiveLocalLogTtlMs(logTtlMs, localLogTtlMs);
- }
-
/** Returns the effective TTL used by local log segment cleanup, in milliseconds. */
public long getEffectiveLocalLogTtlMs() {
return effectiveLocalLogTtlMs;
@@ -1362,12 +1353,14 @@ public final class LogTablet {
}
long now = clock.milliseconds();
+ // TTL can be hot-updated, we need a snapshot of the TTL to ensure consistent
+ long ttlMs = effectiveLocalLogTtlMs;
List<LogSegment> deletableSegments =
- deletableExpiredSegments(endOffset, now, logSegments);
+ deletableExpiredSegments(endOffset, now, logSegments, ttlMs);
if (deletableSegments.size() == logSegments.size() - 1
&& shouldRollExpiredActiveSegment(
- now, logSegments.get(logSegments.size() - 1))) {
+ now, logSegments.get(logSegments.size() - 1), ttlMs)) {
roll(Optional.empty());
}
@@ -1387,6 +1380,7 @@ public final class LogTablet {
// readers is in progress.
List<LogSegment> deletableSegments = new ArrayList<>();
List<LogSegment> logSegments = localLog.getSegments().values();
+ int tieredLogLocalSegments = getTieredLogLocalSegments();
// ignore the segments configured to be retained
for (int i = 0; i < logSegments.size() - tieredLogLocalSegments; i++) {
@@ -1401,11 +1395,11 @@ public final class LogTablet {
/** Returns the contiguous prefix of expired inactive segments eligible for deletion. */
private List<LogSegment> deletableExpiredSegments(
- long endOffset, long now, List<LogSegment> logSegments) throws IOException {
+ long endOffset, long now, List<LogSegment> logSegments, long ttlMs) throws IOException {
List<LogSegment> deletableSegments = new ArrayList<>();
for (int i = 0; i < logSegments.size() - 1; i++) {
if (logSegments.get(i + 1).getBaseOffset() > endOffset
- || !isSegmentExpired(now, logSegments.get(i), effectiveLocalLogTtlMs)) {
+ || !isSegmentExpired(now, logSegments.get(i), ttlMs)) {
break;
}
deletableSegments.add(logSegments.get(i));
@@ -1414,29 +1408,24 @@ public final class LogTablet {
}
/** Returns whether the active segment is non-empty, expired, and fully committed. */
- private boolean shouldRollExpiredActiveSegment(long now, LogSegment activeSegment)
+ private boolean shouldRollExpiredActiveSegment(long now, LogSegment activeSegment, long ttlMs)
throws IOException {
return rollExpiredActiveSegmentEnabled.get()
&& activeSegment.getSizeInBytes() > 0
- && isSegmentExpired(now, activeSegment, effectiveLocalLogTtlMs)
+ && isSegmentExpired(now, activeSegment, ttlMs)
&& getHighWatermark() >= localLogEndOffset();
}
- private long effectiveLocalLogTtlMs(long logTtlMs, long localLogTtlMs) {
- return remoteLogEnabled ? localLogTtlMs : logTtlMs;
- }
-
@FunctionalInterface
private interface SegmentCleanupAction {
void cleanup(long endOffset) throws IOException;
}
- private boolean isSegmentExpired(long now, LogSegment segment, long expirationTimeMs)
- throws IOException {
- if (expirationTimeMs <= 0L) {
+ private boolean isSegmentExpired(long now, LogSegment segment, long ttlMs) throws IOException {
+ if (ttlMs <= 0L) {
return false;
}
- return now - segment.maxTimestampSoFar() > expirationTimeMs;
+ return now - segment.maxTimestampSoFar() > ttlMs;
}
private void deleteSegments(List<LogSegment> deletableSegments, SegmentDeletionReason reason)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java
index 0731ef04b..5e7c57113 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.log.remote;
+import org.apache.fluss.config.TableConfig;
import org.apache.fluss.exception.RemoteStorageException;
import org.apache.fluss.exception.RetriableException;
import org.apache.fluss.fs.FsPath;
@@ -138,10 +139,14 @@ public class LogTieringTask implements Runnable {
List<EnrichedLogSegment> candidateToCopySegments =
candidateToCopyLogSegments(logTablet);
// Only delete segments that have been tiered to lake to ensure data safety
+ TableConfig tableConfig = replica.getTableInfo().getTableConfig();
List<RemoteLogSegment> expiredRemoteLogSegments =
remoteLog.expiredRemoteLogSegments(
clock.milliseconds(),
- logTablet.isDataLakeEnabled() ? logTablet.getLakeLogEndOffset() : null);
+ tableConfig.isDataLakeEnabled()
+ ? logTablet.getLakeLogEndOffset()
+ : null,
+ tableConfig.getLogTTLMs());
// 1. For these candidateToCopySegments, we will first copy segment files to
// remote before commit the remote log manifest.
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 2e6e171c9..e31ceafe1 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
@@ -152,8 +152,7 @@ public class RemoteLogManager implements Closeable {
TableBucket tableBucket = replica.getTableBucket();
PhysicalTablePath physicalTablePath = replica.getPhysicalTablePath();
LogTablet log = replica.getLogTablet();
- RemoteLogTablet remoteLog =
- new RemoteLogTablet(physicalTablePath, tableBucket, replica.getLogTTLMs());
+ RemoteLogTablet remoteLog = new RemoteLogTablet(physicalTablePath, tableBucket);
Optional<RemoteLogManifestHandle> remoteLogManifestHandleOpt =
zkClient.getRemoteLogManifestHandle(tableBucket);
if (remoteLogManifestHandleOpt.isPresent()) {
@@ -290,25 +289,6 @@ public class RemoteLogManager implements Closeable {
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;
}
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 62eb84bc7..4c8f03ba6 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,8 +71,6 @@ public class RemoteLogTablet {
/** The lock to protect the remote log segment list. */
private final ReadWriteLock lock = new ReentrantReadWriteLock();
- private volatile long ttlMs;
-
/** The registered metrics for remote log. */
private volatile MetricGroup remoteLogMetrics;
@@ -94,9 +92,7 @@ public class RemoteLogTablet {
private volatile boolean closed = false;
- public RemoteLogTablet(
- PhysicalTablePath physicalTablePath, TableBucket tableBucket, long ttlMs) {
- this.ttlMs = ttlMs;
+ public RemoteLogTablet(PhysicalTablePath physicalTablePath, TableBucket tableBucket) {
this.currentManifest =
new RemoteLogManifest(physicalTablePath, tableBucket, new ArrayList<>());
reset();
@@ -157,15 +153,12 @@ public class RemoteLogTablet {
* @param currentTimeMs the current time in milliseconds
* @param lakeLogEndOffset the log end offset that has been synced to lake, null if data lake is
* disabled
+ * @param ttlMs the current table log TTL in milliseconds
* @return list of expired segments that can be safely deleted
*/
public List<RemoteLogSegment> expiredRemoteLogSegments(
- long currentTimeMs, Long lakeLogEndOffset) {
- // 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) {
+ long currentTimeMs, Long lakeLogEndOffset, long ttlMs) {
+ if (ttlMs <= 0) {
return Collections.emptyList();
}
return inReadLock(
@@ -175,7 +168,7 @@ public class RemoteLogTablet {
for (Map.Entry<Long, Set<UUID>> entry :
timestampToRemoteLogSegmentId.entrySet()) {
long ts = entry.getKey();
- if (currentTimeMs - ts > ttlSnapshotMs) {
+ if (currentTimeMs - ts > ttlMs) {
for (UUID uuid : entry.getValue()) {
RemoteLogSegment segment = idToRemoteLogSegment.get(uuid);
if (lakeLogEndOffset != null) {
@@ -340,20 +333,6 @@ public class RemoteLogTablet {
.add(remoteLogSegmentId);
}
- /** 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() {
idToRemoteLogSegment.clear();
offsetToRemoteLogSegmentId.clear();
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 47eb0e7b7..0d30b8381 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
@@ -20,6 +20,7 @@ package org.apache.fluss.server.replica;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.compression.ArrowCompressionInfo;
import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.ConfigurationUtils;
import org.apache.fluss.config.TableConfig;
import org.apache.fluss.exception.FencedLeaderEpochException;
import org.apache.fluss.exception.InvalidColumnProjectionException;
@@ -184,15 +185,10 @@ public final class Replica {
private final AdjustIsrManager adjustIsrManager;
private final SchemaGetter schemaGetter;
- private final TableInfo tableInfo;
- private final TableConfig tableConfig;
- // logFormat and arrowCompressionInfo are used in hot-path, so cache them here.
+ private volatile TableInfo tableInfo;
+ // logFormat and arrowCompressionInfo are immutable and 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;
@@ -275,16 +271,14 @@ public final class Replica {
tableInfo.getSchemaId(),
tableInfo.getSchema());
this.tableInfo = tableInfo;
- this.tableConfig = tableInfo.getTableConfig();
+ TableConfig 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();
this.logTablet = createLog(dataDir, lazyHighWatermarkCheckpoint);
- this.logTablet.updateIsDataLakeEnabled(tableConfig.isDataLakeEnabled());
this.clock = clock;
this.remoteLogManager = remoteLogManager;
this.scannerManager = checkNotNull(scannerManager, "scannerManager");
@@ -350,6 +344,10 @@ public final class Replica {
return tableInfo;
}
+ private TableConfig getTableConfig() {
+ return tableInfo.getTableConfig();
+ }
+
public @Nullable Integer getLeaderId() {
return leaderReplicaIdOpt.get();
}
@@ -367,7 +365,7 @@ public final class Replica {
}
public boolean isDataLakeEnabled() {
- return tableConfig.isDataLakeEnabled();
+ return getTableConfig().isDataLakeEnabled();
}
public long getLocalLogStartOffset() {
@@ -386,10 +384,6 @@ public final class Replica {
return tableBucket;
}
- public long getLogTTLMs() {
- return latestLogTtlMs;
- }
-
public int writerIdCount() {
return logTablet.getWriterIdCount();
}
@@ -660,84 +654,46 @@ public final class Replica {
logTablet.updateLeaderEndOffsetSnapshot();
}
- public void updateIsDataLakeEnabled(boolean isDataLakeEnabled) {
- boolean old = logTablet.isDataLakeEnabled();
- if (old == isDataLakeEnabled) {
- return;
- }
-
- logTablet.updateIsDataLakeEnabled(isDataLakeEnabled);
+ /** Updates the table metadata snapshot used by this replica. */
+ public void updateTableInfo(TableInfo tableInfo) {
+ TableInfo oldTableInfo = this.tableInfo;
+ TableInfo newTableInfo = checkNotNull(tableInfo, "tableInfo");
+ boolean wasDataLakeEnabled = isDataLakeEnabled();
+ this.tableInfo = newTableInfo;
+ logTablet.applyTableConfig(newTableInfo.getTableConfig());
- if (isLeader()) {
+ boolean isDataLakeEnabled = isDataLakeEnabled();
+ if (isLeader() && wasDataLakeEnabled != isDataLakeEnabled) {
if (isDataLakeEnabled) {
registerLakeTieringMetrics();
- } else {
- if (lakeTieringMetricGroup != null) {
- lakeTieringMetricGroup.close();
- lakeTieringMetricGroup = null;
- }
+ } else if (lakeTieringMetricGroup != null) {
+ lakeTieringMetricGroup.close();
+ lakeTieringMetricGroup = null;
}
}
- LOG.info(
- "Replica for {} isDataLakeEnabled changed from {} to {}",
- tableBucket,
- old,
- isDataLakeEnabled);
- }
-
- /**
- * Update the number of log segments to retain in local storage. This method is called when the
- * table configuration is altered.
- *
- * @param tieredLogLocalSegments the new number of segments to retain locally
- */
- public void updateTieredLogLocalSegments(int tieredLogLocalSegments) {
- int oldValue = logTablet.getTieredLogLocalSegments();
- if (oldValue == tieredLogLocalSegments) {
- return;
- }
-
- logTablet.updateTieredLogLocalSegments(tieredLogLocalSegments);
-
- LOG.info(
- "Replica for {} tieredLogLocalSegments changed from {} to {}",
- tableBucket,
- oldValue,
- tieredLogLocalSegments);
+ logTableConfigChanges(oldTableInfo, newTableInfo);
}
- /**
- * Updates the log TTLs after the table configuration 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.
- */
- public void updateLogTtls(long logTtlMs, long localLogTtlMs) {
- latestLogTtlMs = logTtlMs;
-
- long oldEffectiveTtl = logTablet.getEffectiveLocalLogTtlMs();
- logTablet.updateLogTtls(logTtlMs, localLogTtlMs);
- long newEffectiveTtl = logTablet.getEffectiveLocalLogTtlMs();
-
- Optional<Long> remoteOldValueOpt = remoteLogManager.updateLogTtlMs(tableBucket, logTtlMs);
- if (!remoteOldValueOpt.isPresent()) {
- LOG.debug(
- "RemoteLogTablet for {} is unavailable; cached new logTtlMs={} "
- + "(remote logging may be disabled or the replica is still initializing).",
- tableBucket,
- logTtlMs);
- return;
- }
+ private void logTableConfigChanges(TableInfo oldTableInfo, TableInfo newTableInfo) {
+ Map<String, String> oldProperties = oldTableInfo.getProperties().toMap();
+ Map<String, String> newProperties = newTableInfo.getProperties().toMap();
+ Set<String> configKeys = new HashSet<>(oldProperties.keySet());
+ configKeys.addAll(newProperties.keySet());
- if (oldEffectiveTtl == newEffectiveTtl) {
- return;
+ for (String configKey : configKeys) {
+ String oldValue = oldProperties.get(configKey);
+ String newValue = newProperties.get(configKey);
+ if (!Objects.equals(oldValue, newValue)) {
+ LOG.info(
+ "Table config {} for bucket {} of table {} changed from {} to {}.",
+ configKey,
+ tableBucket,
+ physicalPath,
+ ConfigurationUtils.hideSensitiveValue(configKey, oldValue),
+ ConfigurationUtils.hideSensitiveValue(configKey, newValue));
+ }
}
-
- LOG.info(
- "Replica for {} effectiveLocalLogTtlMs changed from {} to {}",
- tableBucket,
- oldEffectiveTtl,
- newEffectiveTtl);
}
private void createKv() {
@@ -822,6 +778,7 @@ public final class Replica {
*/
private Optional<CompletedSnapshot> initKvTablet() {
checkNotNull(kvManager);
+ TableConfig tableConfig = getTableConfig();
long startTime = clock.milliseconds();
LOG.info("Start to init kv tablet for {} of table {}.", tableBucket, physicalPath);
@@ -989,6 +946,7 @@ public final class Replica {
long startRecoverLogOffset,
@Nullable Long rowCount,
@Nullable AutoIncIDRange autoIncIDRange) {
+ TableConfig tableConfig = getTableConfig();
long start = clock.milliseconds();
checkNotNull(kvTablet, "kv tablet should not be null.");
try {
@@ -2298,7 +2256,7 @@ public final class Replica {
public boolean isUnderReplicated() {
// is leader and isr size less than numReplicas
- return isLeader() && isrState.isr().size() < tableConfig.getReplicationFactor();
+ return isLeader() && isrState.isr().size() < getTableConfig().getReplicationFactor();
}
public boolean isUnderMinIsr() {
@@ -2326,16 +2284,11 @@ public final class Replica {
private LogTablet createLog(
File dataDir, OffsetCheckpointFile.LazyOffsetCheckpoints lazyHighWatermarkCheckpoint)
throws Exception {
+ TableConfig tableConfig = getTableConfig();
LogTablet log =
logManager.getOrCreateLog(
- dataDir,
- physicalPath,
- tableBucket,
- tableConfig.getLogFormat(),
- tableConfig.getTieredLogLocalSegments(),
- tableConfig.getLogTTLMs(),
- tableConfig.getLocalLogTTLMs(),
- isKvTable());
+ dataDir, physicalPath, tableBucket, tableConfig, isKvTable());
+ log.applyTableConfig(tableConfig);
// update high watermark.
Optional<Long> watermarkOpt = lazyHighWatermarkCheckpoint.fetch(tableBucket);
long watermark =
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 a4fceb141..db8b2e8ed 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
@@ -612,7 +612,7 @@ public class ReplicaManager implements ServerReconfigurable {
validateAndApplyCoordinatorEpoch(coordinatorEpoch, "updateMetadataCache");
Set<Long> deletedTableIds =
metadataCache.updateClusterMetadata(clusterMetadata);
- updateReplicaTableConfig(clusterMetadata);
+ updateReplicaTableInfo(clusterMetadata);
if (!deletedTableIds.isEmpty()) {
ioExecutor.execute(
() ->
@@ -623,31 +623,13 @@ public class ReplicaManager implements ServerReconfigurable {
});
}
- private void updateReplicaTableConfig(ClusterMetadata clusterMetadata) {
- Map<Long, Boolean> tableIdToLakeFlag = new HashMap<>();
- Map<Long, Integer> tableIdToTieredLogLocalSegments = new HashMap<>();
- Map<Long, TableConfig> tableIdToTableConfig = new HashMap<>();
-
+ private void updateReplicaTableInfo(ClusterMetadata clusterMetadata) {
+ Map<Long, TableInfo> tableInfosById = new HashMap<>();
for (TableMetadata tableMetadata : clusterMetadata.getTableMetadataList()) {
TableInfo tableInfo = tableMetadata.getTableInfo();
- long tableId = tableInfo.getTableId();
-
- // Collect datalake enabled configuration
- if (tableInfo.getTableConfig().getDataLakeFormat().isPresent()) {
- boolean dataLakeEnabled = tableInfo.getTableConfig().isDataLakeEnabled();
- tableIdToLakeFlag.put(tableId, dataLakeEnabled);
- }
-
- // Collect tiered log local segments configuration
- int tieredLogLocalSegments = tableInfo.getTableConfig().getTieredLogLocalSegments();
- tableIdToTieredLogLocalSegments.put(tableId, tieredLogLocalSegments);
-
- tableIdToTableConfig.put(tableId, tableInfo.getTableConfig());
+ tableInfosById.put(tableInfo.getTableId(), tableInfo);
}
-
- if (tableIdToLakeFlag.isEmpty()
- && tableIdToTieredLogLocalSegments.isEmpty()
- && tableIdToTableConfig.isEmpty()) {
+ if (tableInfosById.isEmpty()) {
return;
}
@@ -656,23 +638,9 @@ public class ReplicaManager implements ServerReconfigurable {
if (hostedReplica instanceof OnlineReplica) {
Replica replica = ((OnlineReplica) hostedReplica).getReplica();
long tableId = replica.getTableBucket().getTableId();
-
- // Update datalake enabled configuration
- if (tableIdToLakeFlag.containsKey(tableId)) {
- replica.updateIsDataLakeEnabled(tableIdToLakeFlag.get(tableId));
- }
-
- // Update tiered log local segments configuration
- if (tableIdToTieredLogLocalSegments.containsKey(tableId)) {
- replica.updateTieredLogLocalSegments(
- tableIdToTieredLogLocalSegments.get(tableId));
- }
-
- // Update log TTL configurations
- if (tableIdToTableConfig.containsKey(tableId)) {
- TableConfig tableConfig = tableIdToTableConfig.get(tableId);
- replica.updateLogTtls(
- tableConfig.getLogTTLMs(), tableConfig.getLocalLogTTLMs());
+ TableInfo tableInfo = tableInfosById.get(tableId);
+ if (tableInfo != null) {
+ replica.updateTableInfo(tableInfo);
}
}
}
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 68507a170..6d043ed16 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
@@ -29,6 +29,8 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
@@ -123,9 +125,13 @@ final class LocalSegmentTTLTest extends ReplicaTestBase {
void testLogTtlRemainsEffectiveWhenRemoteLogDisabled() throws Exception {
TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0);
makeLogTableAsLeader(tableBucket, false);
- LogTablet logTablet = replicaManager.getReplicaOrException(tableBucket).getLogTablet();
+ Replica replica = replicaManager.getReplicaOrException(tableBucket);
+ LogTablet logTablet = replica.getLogTablet();
- logTablet.updateLogTtls(Duration.ofHours(2).toMillis(), Duration.ofMinutes(30).toMillis());
+ Map<String, String> configUpdates = new HashMap<>();
+ configUpdates.put(ConfigOptions.TABLE_LOG_TTL.key(), "2h");
+ configUpdates.put(ConfigOptions.TABLE_LOG_LOCAL_TTL.key(), "30m");
+ updateTableConfig(replica, configUpdates);
assertThat(logTablet.getEffectiveLocalLogTtlMs()).isEqualTo(Duration.ofHours(2).toMillis());
}
@@ -144,7 +150,8 @@ final class LocalSegmentTTLTest extends ReplicaTestBase {
Replica replica = replicaManager.getReplicaOrException(tb);
LogTablet logTablet = replica.getLogTablet();
- // Remote log is disabled in this test base, so updateLogTtls must still reach LogTablet.
+ // Remote log is disabled in this test base, so LogTablet must still read the updated
+ // TableInfo.
assertThatThrownBy(() -> remoteLogManager.remoteLogTablet(tb))
.isInstanceOf(IllegalStateException.class);
@@ -156,7 +163,7 @@ final class LocalSegmentTTLTest extends ReplicaTestBase {
assertThat(logTablet.getSegments()).hasSize(1);
// Shrink TTL to 1ms; the segment should now be expired.
- replica.updateLogTtls(1L, 1L);
+ updateTableConfig(replica, ConfigOptions.TABLE_LOG_TTL, "1ms");
logManager.cleanupExpiredLocalLogSegments();
// The expired active segment is rolled, creating a new empty segment at offset 10.
assertThat(logTablet.getSegments()).hasSize(2);
@@ -168,7 +175,7 @@ final class LocalSegmentTTLTest extends ReplicaTestBase {
assertThat(logTablet.localLogStartOffset()).isEqualTo(10L);
// Disable expiration; the remaining segment must survive even after a long time.
- replica.updateLogTtls(-1L, -1L);
+ updateTableConfig(replica, ConfigOptions.TABLE_LOG_TTL, "0ms");
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 514dd8f77..5ac61b0bb 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
@@ -444,14 +444,16 @@ public class RemoteLogITCase {
false))
.get();
- // Wait until the new TTL reaches the leader's remote and local tablets and the follower's
- // local tablet.
+ // Wait until the new TableInfo reaches the leader and follower. Both local and remote
+ // retention read the TTL from this metadata snapshot.
retry(
Duration.ofMinutes(1),
() -> {
- assertThat(remoteLogTablet.getTtlMs()).isEqualTo(newTtlMs);
+ assertThat(leaderReplica.getTableInfo().getTableConfig().getLogTTLMs())
+ .isEqualTo(newTtlMs);
assertThat(leaderLogTablet.getEffectiveLocalLogTtlMs()).isEqualTo(newTtlMs);
- assertThat(followerReplica.getLogTTLMs()).isEqualTo(newTtlMs);
+ assertThat(followerReplica.getTableInfo().getTableConfig().getLogTTLMs())
+ .isEqualTo(newTtlMs);
assertThat(followerLogTablet.getEffectiveLocalLogTtlMs()).isEqualTo(newTtlMs);
});
@@ -487,10 +489,12 @@ public class RemoteLogITCase {
.isGreaterThan(initialFollowerLocalStartOffset);
});
- // Verify follower caching still preserves the shortened TTL through leader promotion.
+ // Verify the follower preserves the latest TableInfo through leader promotion.
retry(
Duration.ofMinutes(1),
- () -> assertThat(followerReplica.getLogTTLMs()).isEqualTo(newTtlMs));
+ () ->
+ assertThat(followerReplica.getTableInfo().getTableConfig().getLogTTLMs())
+ .isEqualTo(newTtlMs));
// --- Failover: stop leader, wait for new leader ---
FLUSS_CLUSTER_EXTENSION.stopTabletServer(leaderId);
@@ -498,7 +502,7 @@ public class RemoteLogITCase {
int newLeaderId = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb);
assertThat(newLeaderId).isNotEqualTo(leaderId);
- // The promoted follower must construct its RemoteLogTablet with the altered TTL.
+ // The promoted follower must retain the altered TableInfo and resume remote tiering.
TabletServer newLeaderServer = FLUSS_CLUSTER_EXTENSION.getTabletServerById(newLeaderId);
RemoteLogManager newRemoteLogManager =
newLeaderServer.getReplicaManager().getRemoteLogManager();
@@ -507,7 +511,10 @@ public class RemoteLogITCase {
() -> {
RemoteLogTablet newRemoteLogTablet =
newRemoteLogManager.remoteLogTablet(tb);
- assertThat(newRemoteLogTablet.getTtlMs()).isEqualTo(newTtlMs);
+ Replica newLeaderReplica =
+ newLeaderServer.getReplicaManager().getReplicaOrException(tb);
+ assertThat(newLeaderReplica.getTableInfo().getTableConfig().getLogTTLMs())
+ .isEqualTo(newTtlMs);
assertThat(newRemoteLogTablet.allRemoteLogSegments()).isNotEmpty();
});
} finally {
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 139e762ca..ba876c2e7 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
@@ -765,7 +765,7 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
assertThat(logTablet.getSegments()).hasSize(2);
// 3. Directly update config via Replica (simulating metadata propagation)
- replica.updateTieredLogLocalSegments(5);
+ updateTableConfig(replica, ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, "5");
// Verify LogTablet internal state has been updated
assertThat(logTablet.getTieredLogLocalSegments()).isEqualTo(5);
@@ -778,7 +778,7 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
assertThat(logTablet.getSegments()).hasSize(5);
// 5. Modify config to 3 again, verify multiple modifications work
- replica.updateTieredLogLocalSegments(3);
+ updateTableConfig(replica, ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, "3");
// Verify LogTablet internal state updated again
assertThat(logTablet.getTieredLogLocalSegments()).isEqualTo(3);
@@ -793,7 +793,7 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
@ParameterizedTest
@ValueSource(booleans = {true, false})
- void testUpdateLogTtlMsDelegation(boolean partitionedTable) throws Exception {
+ void testUpdateTableInfo(boolean partitionedTable) throws Exception {
long tableId =
registerTableInZkClient(
DATA1_TABLE_PATH,
@@ -805,48 +805,20 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
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);
+ assertThat(replica.getTableInfo().getTableConfig().getLogTTLMs()).isEqualTo(defaultTtlMs);
- // Update ttl and verify RemoteLogTablet reflects the new value.
+ // Update the complete TableInfo and verify all config consumers see the new value.
long newTtlMs = Duration.ofDays(1).toMillis();
- replica.updateLogTtls(newTtlMs, newTtlMs);
- assertThat(remoteLog.getTtlMs()).isEqualTo(newTtlMs);
+ updateTableConfig(replica, ConfigOptions.TABLE_LOG_TTL, "1d");
+ assertThat(replica.getTableInfo().getTableConfig().getLogTTLMs()).isEqualTo(newTtlMs);
+ assertThat(replica.getLogTablet().getEffectiveLocalLogTtlMs()).isEqualTo(newTtlMs);
// no-op: same value must not break anything.
- replica.updateLogTtls(newTtlMs, 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.updateLogTtls(newTtlMs, newTtlMs);
- assertThat(replica.getLogTTLMs()).isEqualTo(newTtlMs);
-
- // Simulate promotion: registerReplica must use the cached TTL.
- remoteLogManager.registerReplica(replica);
- assertThat(remoteLogManager.remoteLogTablet(tb).getTtlMs()).isEqualTo(newTtlMs);
+ updateTableConfig(replica, ConfigOptions.TABLE_LOG_TTL, "1d");
+ assertThat(replica.getTableInfo().getTableConfig().getLogTTLMs()).isEqualTo(newTtlMs);
}
@ParameterizedTest
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java
index 97aed3f76..5b5971880 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTTLTest.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.log.remote;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
@@ -25,6 +26,7 @@ import org.apache.fluss.server.entity.FetchReqInfo;
import org.apache.fluss.server.log.FetchParams;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.log.remote.RemoteLogIndexCache.Entry;
+import org.apache.fluss.server.replica.Replica;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.params.ParameterizedTest;
@@ -57,9 +59,10 @@ final class RemoteLogTTLTest extends RemoteLogTestBase {
}
// Need to make leader by ReplicaManager.
makeLogTableAsLeader(tb, partitionTable);
- LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
// enable data lake
- logTablet.updateIsDataLakeEnabled(true);
+ updateTableConfig(replica, ConfigOptions.TABLE_DATALAKE_ENABLED, "true");
// add 5 segments, so 4 segments will be uploaded to remote (exclude active segment)
// segment offsets: [0,10), [10,20), [20,30), [30,40), [40,50) active
addMultiSegmentsToLogTablet(logTablet, 5);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletOverlapTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletOverlapTest.java
index d9cfb7ee1..de1afe674 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletOverlapTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletOverlapTest.java
@@ -41,7 +41,7 @@ class RemoteLogTabletOverlapTest {
void testLogicalLookupUsesClippedRanges() {
RemoteLogSegment first = segment(0L, 10L).withLogicalRange(0L, 5L);
RemoteLogSegment second = segment(5L, 20L);
- RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET, -1L);
+ RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET);
tablet.loadRemoteLogManifest(
new RemoteLogManifest(TABLE_PATH, TABLE_BUCKET, Arrays.asList(first, second)));
@@ -55,7 +55,7 @@ class RemoteLogTabletOverlapTest {
void testFetchV0StopsBeforePhysicalOverlap() {
RemoteLogSegment first = segment(0L, 10L).withLogicalRange(0L, 5L);
RemoteLogSegment second = segment(5L, 20L);
- RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET, -1L);
+ RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET);
tablet.loadRemoteLogManifest(
new RemoteLogManifest(TABLE_PATH, TABLE_BUCKET, Arrays.asList(first, second)));
@@ -67,7 +67,7 @@ class RemoteLogTabletOverlapTest {
void testFetchV0ReturnsContiguousSegmentsTogether() {
List<RemoteLogSegment> segments =
Arrays.asList(segment(0L, 10L), segment(10L, 20L), segment(20L, 30L));
- RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET, -1L);
+ RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET);
tablet.loadRemoteLogManifest(new RemoteLogManifest(TABLE_PATH, TABLE_BUCKET, segments));
assertThat(tablet.relevantRemoteLogSegmentsForFetchV0(5L))
@@ -76,7 +76,7 @@ class RemoteLogTabletOverlapTest {
@Test
void testEmptyManifestKeepsCopyProgress() {
- RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET, -1L);
+ RemoteLogTablet tablet = new RemoteLogTablet(TABLE_PATH, TABLE_BUCKET);
tablet.loadRemoteLogManifest(
new RemoteLogManifest(TABLE_PATH, TABLE_BUCKET, Collections.emptyList(), 20L));
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 1f9d7ea58..e57d2794e 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
@@ -152,13 +152,12 @@ class RemoteLogTabletTest extends RemoteLogTestBase {
@ParameterizedTest
@ValueSource(booleans = {true, false})
- void testUpdateTtlMs(boolean partitionTable) throws Exception {
+ void testExpireSegmentsWithCurrentTtl(boolean partitionTable) throws Exception {
LogTablet logTablet = makeLogTabletAndAddSegments(partitionTable);
RemoteLogTablet remoteLogTablet = buildRemoteLogTablet(logTablet);
- // initial ttl follows the default in ConfigOptions.TABLE_LOG_TTL (7 days)
+ // The current table TTL is supplied by the tiering task for each expiration pass.
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);
@@ -166,18 +165,15 @@ class RemoteLogTabletTest extends RemoteLogTestBase {
// 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();
+ assertThat(remoteLogTablet.expiredRemoteLogSegments(oneHourMs, null, defaultTtlMs))
+ .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))
+ // With a 1 ms TTL, the same segment is expired.
+ assertThat(remoteLogTablet.expiredRemoteLogSegments(oneHourMs, null, 1L))
.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();
+ // A non-positive current TTL disables expiration.
+ assertThat(remoteLogTablet.expiredRemoteLogSegments(oneHourMs, null, -1L)).isEmpty();
}
RemoteLogSegment createLogSegmentWithMaxTimestamp(
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java
index ef6b2875d..a533c3ecd 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java
@@ -133,10 +133,7 @@ public class RemoteLogTestBase extends ReplicaTestBase {
}
protected RemoteLogTablet buildRemoteLogTablet(LogTablet logTablet) {
- return new RemoteLogTablet(
- logTablet.getPhysicalTablePath(),
- logTablet.getTableBucket(),
- conf.get(ConfigOptions.TABLE_LOG_TTL).toMillis());
+ return new RemoteLogTablet(logTablet.getPhysicalTablePath(), logTablet.getTableBucket());
}
protected static List<RemoteLogSegment> createRemoteLogSegmentList(LogTablet logTablet) {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java
index 6e09f9384..66d687e7d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TieredLocalSegmentTTLTest.java
@@ -21,6 +21,7 @@ import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.server.log.LogSegment;
import org.apache.fluss.server.log.LogTablet;
+import org.apache.fluss.server.replica.Replica;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -185,10 +186,11 @@ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase {
conf.set(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED, true);
logManager.reconfigure(conf);
makeLogTableAsLeader(tb, partitionTable);
- LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
addMultiSegmentsToLogTablet(logTablet, 5);
- logTablet.updateTieredLogLocalSegments(5);
+ updateTableConfig(replica, ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, "5");
logTablet.updateHighestCopiedEndOffset(20L);
manualClock.advanceTime(Duration.ofMinutes(90));
@@ -210,19 +212,20 @@ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase {
void testUpdatedLocalTtlTakesEffectImmediately() throws Exception {
TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0);
makeLogTableAsLeader(tableBucket, false);
- LogTablet logTablet = replicaManager.getReplicaOrException(tableBucket).getLogTablet();
- logTablet.updateTieredLogLocalSegments(5);
+ Replica replica = replicaManager.getReplicaOrException(tableBucket);
+ LogTablet logTablet = replica.getLogTablet();
+ updateTableConfig(replica, ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, "5");
addMultiSegmentsToLogTablet(logTablet, 5);
remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
- logTablet.updateLogTtls(Duration.ofHours(2).toMillis(), Duration.ofHours(2).toMillis());
+ updateTableConfig(replica, ConfigOptions.TABLE_LOG_LOCAL_TTL, "2h");
assertThat(logTablet.getEffectiveLocalLogTtlMs()).isEqualTo(Duration.ofHours(2).toMillis());
manualClock.advanceTime(Duration.ofMinutes(90));
logManager.cleanupExpiredLocalLogSegments();
assertThat(logTablet.getSegments()).hasSize(5);
- logTablet.updateLogTtls(Duration.ofHours(2).toMillis(), Duration.ofHours(1).toMillis());
+ updateTableConfig(replica, ConfigOptions.TABLE_LOG_LOCAL_TTL, "1h");
logManager.cleanupExpiredLocalLogSegments();
assertThat(logTablet.getSegments()).hasSize(1);
assertThat(logTablet.localLogStartOffset()).isEqualTo(40L);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java
index 8b5795644..5ac781b73 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.replica;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.rpc.messages.NotifyLakeTableOffsetResponse;
@@ -62,7 +63,7 @@ class NotifyReplicaLakeTableOffsetTest extends ReplicaTestBase {
TableBucket tb = makeTableBucket(false);
makeLogTableAsLeader(tb, false);
Replica replica = replicaManager.getReplicaOrException(tb);
- replica.updateIsDataLakeEnabled(true);
+ updateTableConfig(replica, ConfigOptions.TABLE_DATALAKE_ENABLED, "true");
long initialTimestamp = manualClock.milliseconds();
MemoryLogRecords records =
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
index 2d8b88830..dacfa846a 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
@@ -1025,7 +1025,7 @@ final class ReplicaTest extends ReplicaTestBase {
Replica logReplica =
makeLogReplica(DATA1_PHYSICAL_TABLE_PATH, new TableBucket(DATA1_TABLE_ID, 1));
makeLogReplicaAsLeader(logReplica);
- logReplica.updateIsDataLakeEnabled(true);
+ updateTableConfig(logReplica, ConfigOptions.TABLE_DATALAKE_ENABLED, "true");
long initialTimestamp = manualClock.milliseconds();
MemoryLogRecords firstBatch =
@@ -1072,7 +1072,7 @@ final class ReplicaTest extends ReplicaTestBase {
}
@Test
- void testUpdateIsDataLakeEnabled() throws Exception {
+ void testUpdateTableInfoChangesDataLakeEnabled() throws Exception {
Replica logReplica =
makeLogReplica(DATA1_PHYSICAL_TABLE_PATH, new TableBucket(DATA1_TABLE_ID, 1));
makeLogReplicaAsLeader(logReplica);
@@ -1081,15 +1081,15 @@ final class ReplicaTest extends ReplicaTestBase {
assertThat(logReplica.getLogTablet().isDataLakeEnabled()).isFalse();
// update to true
- logReplica.updateIsDataLakeEnabled(true);
+ updateTableConfig(logReplica, ConfigOptions.TABLE_DATALAKE_ENABLED, "true");
assertThat(logReplica.getLogTablet().isDataLakeEnabled()).isTrue();
// update with same value should not change anything (no-op)
- logReplica.updateIsDataLakeEnabled(true);
+ updateTableConfig(logReplica, ConfigOptions.TABLE_DATALAKE_ENABLED, "true");
assertThat(logReplica.getLogTablet().isDataLakeEnabled()).isTrue();
// update to false
- logReplica.updateIsDataLakeEnabled(false);
+ updateTableConfig(logReplica, ConfigOptions.TABLE_DATALAKE_ENABLED, "false");
assertThat(logReplica.getLogTablet().isDataLakeEnabled()).isFalse();
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
index 400d64460..2159e7ca2 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
@@ -19,6 +19,7 @@ package org.apache.fluss.server.replica;
import org.apache.fluss.cluster.Endpoint;
import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.ConfigOption;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.MemorySize;
@@ -27,6 +28,7 @@ import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.rpc.RpcClient;
@@ -85,6 +87,7 @@ import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -501,6 +504,27 @@ public class ReplicaTestBase {
replicaManager.becomeLeaderOrFollower(0, notifyLeaderAndIsrDataList, result -> {});
}
+ protected void updateTableConfig(
+ Replica replica, ConfigOption<?> configOption, String newConfigValue) {
+ updateTableConfig(replica, Collections.singletonMap(configOption.key(), newConfigValue));
+ }
+
+ protected void updateTableConfig(Replica replica, Map<String, String> configUpdates) {
+ TableInfo currentTableInfo = replica.getTableInfo();
+ Map<String, String> properties =
+ new HashMap<>(currentTableInfo.toTableDescriptor().getProperties());
+ properties.putAll(configUpdates);
+ replica.updateTableInfo(
+ TableInfo.of(
+ currentTableInfo.getTablePath(),
+ currentTableInfo.getTableId(),
+ currentTableInfo.getSchemaId(),
+ currentTableInfo.toTableDescriptor().withProperties(properties),
+ currentTableInfo.getRemoteDataDir(),
+ currentTableInfo.getCreatedTime(),
+ currentTableInfo.getModifiedTime()));
+ }
+
protected Replica makeLogReplica(PhysicalTablePath physicalTablePath, TableBucket tableBucket)
throws Exception {
return makeReplica(physicalTablePath, tableBucket, false, null);
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
index 1973f5270..f386a5beb 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -66,8 +66,8 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties)
| auto-increment.fields | String | (None) | Defines the auto increment columns. The auto increment column can only be used in primary-key table. With an auto increment column in the table, whenever a new row is inserted into the table, the new row will be assigned with the next available value from the auto-increment sequence. The data type of the auto increment column must be INT or BIGINT. Currently a table can have only one auto-increme [...]
| bucket.num | int | The bucket number of Fluss cluster. | The number of buckets of a Fluss table. [...]
| bucket.key | String | (None) | Specific the distribution policy of the Fluss table. Data will be distributed to each bucket according to the hash value of bucket-key (It must be a subset of the primary keys excluding partition keys of the primary key table). If you specify multiple fields, delimiter is `,`. If the table has a primary key and a bucket key is not specified, the bucket key will be used as primary key(excluding th [...]
-| table.log.ttl | Duration | 7 days | The time to live for log segments. The configuration controls the maximum time log segments are retained before they become eligible for deletion. When remote log tiering is enabled, this value controls the retention of remote log segments. Setting the value to -1 disables TTL-based deletion. The default value is 7 days. [...]
-| table.log.local-ttl | Duration | (None) | The time to live for local log segments. The configuration controls the maximum time local log segments are retained before they become eligible for deletion. When remote log tiering is enabled, an expired local segment is deleted only after it has been copied to remote storage. Setting the value to -1 disables TTL-based deletion. If not configured, the value inherits `table.log.ttl`. When both v [...]
+| table.log.ttl | Duration | 7 days | The time to live for log segments. The configuration controls the maximum time log segments are retained before they become eligible for deletion. When remote log tiering is enabled, this value controls the retention of remote log segments. Setting the value to '0ms' disables TTL-based deletion. The default value is 7 days. [...]
+| table.log.local-ttl | Duration | (None) | The time to live for local log segments. The configuration controls the maximum time local log segments are retained before they become eligible for deletion. When remote log tiering is enabled, an expired local segment is deleted only after it has been copied to remote storage. Setting the value to '0ms' disables TTL-based deletion. If not configured, the value inherits `table.log.ttl`. When bot [...]
| table.auto-partition.enabled | Boolean | false | Whether enable auto partition for the table. Disable by default. When auto partition is enabled, the partitions of the table will be created automatically. [...]
| table.auto-partition.key | String | (None) | This configuration defines the time-based partition key to be used for auto-partitioning when a table is partitioned with multiple keys. Auto-partitioning utilizes a time-based partition key to handle partitions automatically, including creating new ones and removing outdated ones, by comparing the time value of the partition with the current system time. In the case of a table using multiple par [...]
| table.auto-partition.time-unit | ENUM | DAY | The time granularity for auto created partitions. The default value is `DAY`. Valid values are `HOUR`, `DAY`, `MONTH`, `QUARTER`, `YEAR`. If the value is `HOUR`, the partition format for auto created is yyyyMMddHH. If the value is `DAY`, the partition format for auto created is yyyyMMdd. If the value is `MONTH`, the partition format for auto created is yyyyMM. If the value is `QUARTER`, the parti [...]