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/11 03:18:52 UTC
(fluss) branch main updated: [server] Support hot update for active segment retention roll (#3940)
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
The following commit(s) were added to refs/heads/main by this push:
new 5b58834346 [server] Support hot update for active segment retention roll (#3940)
5b58834346 is described below
commit 5b588343462fb0cb1bfd6e016b0c3c5d06059430
Author: Jark Wu <ja...@apache.org>
AuthorDate: Tue Aug 11 11:18:47 2026 +0800
[server] Support hot update for active segment retention roll (#3940)
Separate tiered and TTL cleanup paths so each retention action owns segment deletion and active-segment rolling. Make log.retention.roll-active-segment.enabled dynamically reconfigurable through a shared AtomicBoolean owned by LogManager, avoiding per-tablet update traversal. Update dynamic configuration and TTL retention coverage.
---
.../apache/fluss/server/DynamicServerConfig.java | 2 +
.../org/apache/fluss/server/log/LogManager.java | 38 ++++++-
.../org/apache/fluss/server/log/LogTablet.java | 126 ++++++++++-----------
.../apache/fluss/server/tablet/TabletServer.java | 2 +
.../fluss/server/DynamicConfigChangeTest.java | 35 ++++++
.../fluss/server/kv/KvTabletMergeModeTest.java | 4 +
.../server/kv/KvTabletSchemaEvolutionTest.java | 4 +
.../org/apache/fluss/server/kv/KvTabletTest.java | 3 +
.../fluss/server/kv/scan/ScannerManagerTest.java | 3 +
.../fluss/server/log/LocalSegmentTTLTest.java | 1 +
.../org/apache/fluss/server/log/LogLoaderTest.java | 3 +
.../apache/fluss/server/log/LogManagerTest.java | 21 ++++
.../org/apache/fluss/server/log/LogTabletTest.java | 5 +
.../log/remote/TieredLocalSegmentTTLTest.java | 4 +
14 files changed, 184 insertions(+), 67 deletions(-)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
index 04a8ef1099..37ae794f53 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
@@ -50,6 +50,7 @@ import static org.apache.fluss.config.ConfigOptions.KV_LEADER_REPLICA_MEMORY_RES
import static org.apache.fluss.config.ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC;
import static org.apache.fluss.config.ConfigOptions.KV_SNAPSHOT_INTERVAL;
import static org.apache.fluss.config.ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER;
+import static org.apache.fluss.config.ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_STRATEGY;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS;
@@ -74,6 +75,7 @@ class DynamicServerConfig {
new HashSet<>(
Arrays.asList(
DATALAKE_FORMAT.key(),
+ LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED.key(),
LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(),
KV_LEADER_REPLICA_MEMORY_RESERVED.key(),
KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(),
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 f18411cbfa..be47711e4e 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
@@ -21,6 +21,7 @@ import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
import org.apache.fluss.exception.FlussException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.LogStorageException;
@@ -67,6 +68,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
@@ -82,7 +84,7 @@ import static org.apache.fluss.utils.concurrent.LockUtils.inLock;
* log instances.
*/
@ThreadSafe
-public final class LogManager extends TabletManagerBase {
+public final class LogManager extends TabletManagerBase implements ServerReconfigurable {
private static final Logger LOG = LoggerFactory.getLogger(LogManager.class);
@VisibleForTesting
@@ -110,6 +112,7 @@ public final class LogManager extends TabletManagerBase {
private volatile Map<File, OffsetCheckpointFile> recoveryPointCheckpoints;
private volatile ScheduledFuture<?> recoveryPointCheckpointTask;
private volatile ScheduledFuture<?> localLogRetentionTask;
+ private final AtomicBoolean rollExpiredActiveSegmentEnabled;
private boolean loadLogsCompletedFlag = false;
private LogManager(
@@ -127,6 +130,9 @@ public final class LogManager extends TabletManagerBase {
this.clock = clock;
this.serverMetricGroup = serverMetricGroup;
this.localDiskManager = localDiskManager;
+ this.rollExpiredActiveSegmentEnabled =
+ new AtomicBoolean(
+ conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED));
initializeCheckpointMaps();
}
@@ -319,6 +325,7 @@ public final class LogManager extends TabletManagerBase {
tablePath,
tabletDir,
conf,
+ rollExpiredActiveSegmentEnabled,
serverMetricGroup,
0L,
scheduler,
@@ -464,6 +471,7 @@ public final class LogManager extends TabletManagerBase {
physicalTablePath,
tabletDir,
conf,
+ rollExpiredActiveSegmentEnabled,
serverMetricGroup,
logRecoveryPoint,
scheduler,
@@ -564,6 +572,34 @@ public final class LogManager extends TabletManagerBase {
}
}
+ // ============ ServerReconfigurable Implementation ============
+
+ @Override
+ public void validate(Configuration newConfig) {
+ // Type validation is handled by DynamicServerConfig.
+ }
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+ boolean newRollExpiredActiveSegmentEnabled =
+ newConfig.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED);
+ boolean oldRollExpiredActiveSegmentEnabled =
+ rollExpiredActiveSegmentEnabled.getAndSet(newRollExpiredActiveSegmentEnabled);
+ if (newRollExpiredActiveSegmentEnabled == oldRollExpiredActiveSegmentEnabled) {
+ LOG.debug(
+ "{} unchanged: {}",
+ ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED.key(),
+ newRollExpiredActiveSegmentEnabled);
+ return;
+ }
+
+ LOG.info(
+ "{} reconfigured: {} -> {}",
+ ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED.key(),
+ oldRollExpiredActiveSegmentEnabled,
+ newRollExpiredActiveSegmentEnabled);
+ }
+
private void waitForShutdownLogsInDir(
File dataDir, List<LogTablet> logs, CompletableFuture<Void> closingFuture) {
boolean closeSucceeded;
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 322db3a6fe..28dd9a93cc 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
@@ -57,7 +57,6 @@ import javax.annotation.concurrent.ThreadSafe;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
-import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -66,9 +65,10 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
-import static org.apache.fluss.utils.FileUtils.flushFileIfExists;
import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
/* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache
* Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
@@ -109,7 +109,7 @@ public final class LogTablet {
private final Clock clock;
private final boolean isChangeLog;
private final long logTtlMs;
- private final boolean rollExpiredActiveSegmentEnabled;
+ private final AtomicBoolean rollExpiredActiveSegmentEnabled;
@GuardedBy("lock")
private volatile LogOffsetMetadata highWatermarkMetadata;
@@ -146,6 +146,7 @@ public final class LogTablet {
PhysicalTablePath physicalPath,
LocalLog localLog,
Configuration conf,
+ AtomicBoolean rollExpiredActiveSegmentEnabled,
Scheduler scheduler,
WriterStateManager writerStateManager,
LogFormat logFormat,
@@ -164,7 +165,9 @@ public final class LogTablet {
this.highWatermarkMetadata = new LogOffsetMetadata(0L);
this.logTtlMs = logTtlMs;
this.rollExpiredActiveSegmentEnabled =
- conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED);
+ checkNotNull(
+ rollExpiredActiveSegmentEnabled,
+ "rollExpiredActiveSegmentEnabled must not be null");
this.scheduler = scheduler;
// scheduler the writer expiration interval check.
@@ -347,6 +350,7 @@ public final class LogTablet {
PhysicalTablePath tablePath,
File tabletDir,
Configuration conf,
+ AtomicBoolean rollExpiredActiveSegmentEnabled,
TabletServerMetricGroup serverMetricGroup,
long recoveryPoint,
Scheduler scheduler,
@@ -397,6 +401,7 @@ public final class LogTablet {
tablePath,
log,
conf,
+ rollExpiredActiveSegmentEnabled,
scheduler,
writerStateManager,
logFormat,
@@ -412,6 +417,7 @@ public final class LogTablet {
PhysicalTablePath tablePath,
File tabletDir,
Configuration conf,
+ AtomicBoolean rollExpiredActiveSegmentEnabled,
TabletServerMetricGroup serverMetricGroup,
long recoveryPoint,
Scheduler scheduler,
@@ -427,6 +433,7 @@ public final class LogTablet {
tablePath,
tabletDir,
conf,
+ rollExpiredActiveSegmentEnabled,
serverMetricGroup,
recoveryPoint,
scheduler,
@@ -669,6 +676,11 @@ public final class LogTablet {
this.tieredLogLocalSegments = tieredLogLocalSegments;
}
+ @VisibleForTesting
+ boolean isRollExpiredActiveSegmentEnabled() {
+ return rollExpiredActiveSegmentEnabled.get();
+ }
+
public int getTieredLogLocalSegments() {
return tieredLogLocalSegments;
}
@@ -755,10 +767,7 @@ public final class LogTablet {
}
public void deleteSegmentsAlreadyExistsInRemote() {
- deleteSegments(
- highestCopiedEndOffset,
- SegmentDeletionReason.LOG_MOVE_TO_REMOTE,
- this::deletableRemoteSegments);
+ cleanupSegments(highestCopiedEndOffset, this::cleanupTieredSegments);
}
/** Deletes inactive local segments that have expired according to the table log TTL. */
@@ -769,10 +778,7 @@ public final class LogTablet {
// cannot be deleted yet.
long cleanupToOffset =
highestCopiedEndOffset == -1L ? getHighWatermark() : highestCopiedEndOffset;
- deleteSegments(
- cleanupToOffset,
- SegmentDeletionReason.LOG_RETENTION,
- this::deletableExpiredSegments);
+ cleanupSegments(cleanupToOffset, this::cleanupExpiredSegments);
}
/**
@@ -789,40 +795,38 @@ public final class LogTablet {
highWatermark);
}
- private void deleteSegments(
- long cleanUpToOffset,
- SegmentDeletionReason reason,
- DeletableSegmentsFinder deletableSegmentsFinder) {
+ private void cleanupSegments(
+ long requestedCleanupToOffset, SegmentCleanupAction cleanupAction) {
// cache to local variables
long localLogStartOffset = localLog.getLocalLogStartOffset();
- if (cleanUpToOffset < localLogStartOffset) {
+ if (requestedCleanupToOffset < localLogStartOffset) {
LOG.debug(
"Ignore the delete segments action for bucket {} while the input cleanUpToOffset {} "
+ "is smaller than the current localLogStartOffset {}",
getTableBucket(),
- cleanUpToOffset,
+ requestedCleanupToOffset,
localLogStartOffset);
return;
}
- if (cleanUpToOffset > getHighWatermark()) {
+ if (requestedCleanupToOffset > getHighWatermark()) {
LOG.warn(
"Ignore the delete segments action for bucket {} while the input cleanUpToOffset {} "
+ "is larger than the current highWatermark {}",
getTableBucket(),
- cleanUpToOffset,
+ requestedCleanupToOffset,
getHighWatermark());
return;
}
try {
// shouldn't clean up segments that will be used by kv recovery.
- long cleanupToOffset = Math.min(minRetainOffset, cleanUpToOffset);
- deleteOldSegments(cleanupToOffset, reason, deletableSegmentsFinder);
+ long effectiveCleanupToOffset = Math.min(minRetainOffset, requestedCleanupToOffset);
+ cleanupAction.cleanup(effectiveCleanupToOffset);
} catch (IOException e) {
LOG.error(
"Failed to delete the local log segments to cleanUpToOffset {} for table-bucket {}.",
- cleanUpToOffset,
+ requestedCleanupToOffset,
getTableBucket(),
e);
// do not re-throw exception as it is not critical.
@@ -1323,33 +1327,40 @@ public final class LogTablet {
}
}
- private void flushWriterStateSnapshot(Path snapshot) {
- try {
- flushFileIfExists(snapshot);
- } catch (IOException e) {
- throw new LogStorageException(
- String.format(
- "Error while flushing writer state snapshot %s for %s in dir %s",
- snapshot, getTableBucket(), getLogDir().getParent()),
- e);
+ private void cleanupTieredSegments(long endOffset) throws IOException {
+ synchronized (lock) {
+ List<LogSegment> deletableSegments = deletableTieredSegments(endOffset);
+ if (!deletableSegments.isEmpty()) {
+ deleteSegments(deletableSegments, SegmentDeletionReason.LOG_MOVE_TO_REMOTE);
+ }
}
}
- private void deleteOldSegments(
- long endOffset,
- SegmentDeletionReason reason,
- DeletableSegmentsFinder deletableSegmentsFinder)
- throws IOException {
+ private void cleanupExpiredSegments(long endOffset) throws IOException {
synchronized (lock) {
- List<LogSegment> deletableSegments = deletableSegmentsFinder.find(endOffset);
+ List<LogSegment> logSegments = localLog.getSegments().values();
+ if (logSegments.isEmpty()) {
+ return;
+ }
+
+ long now = clock.milliseconds();
+ List<LogSegment> deletableSegments =
+ deletableExpiredSegments(endOffset, now, logSegments);
+
+ if (deletableSegments.size() == logSegments.size() - 1
+ && shouldRollExpiredActiveSegment(
+ now, logSegments.get(logSegments.size() - 1))) {
+ roll(Optional.empty());
+ }
+
if (!deletableSegments.isEmpty()) {
- deleteSegments(deletableSegments, reason);
+ deleteSegments(deletableSegments, SegmentDeletionReason.LOG_RETENTION);
}
}
}
/** Returns uploaded segments that exceed the configured local segment retention count. */
- private List<LogSegment> deletableRemoteSegments(long endOffset) {
+ private List<LogSegment> deletableTieredSegments(long endOffset) {
if (localLog.getSegments().isEmpty()) {
return Collections.emptyList();
}
@@ -1370,21 +1381,10 @@ public final class LogTablet {
return deletableSegments;
}
- /**
- * Returns the contiguous prefix of expired inactive segments.
- *
- * <p>If all inactive segments are deletable, this also checks whether the active segment should
- * be rolled.
- */
- private List<LogSegment> deletableExpiredSegments(long endOffset) throws IOException {
- if (localLog.getSegments().isEmpty()) {
- return Collections.emptyList();
- }
-
+ /** Returns the contiguous prefix of expired inactive segments eligible for deletion. */
+ private List<LogSegment> deletableExpiredSegments(
+ long endOffset, long now, List<LogSegment> logSegments) throws IOException {
List<LogSegment> deletableSegments = new ArrayList<>();
- List<LogSegment> logSegments = localLog.getSegments().values();
- long now = clock.milliseconds();
-
for (int i = 0; i < logSegments.size() - 1; i++) {
if (logSegments.get(i + 1).getBaseOffset() > endOffset
|| !isSegmentExpired(now, logSegments.get(i), logTtlMs)) {
@@ -1392,27 +1392,21 @@ public final class LogTablet {
}
deletableSegments.add(logSegments.get(i));
}
-
- if (deletableSegments.size() == logSegments.size() - 1) {
- maybeRollExpiredActiveSegment(now, logSegments.get(logSegments.size() - 1));
- }
return deletableSegments;
}
- /** Rolls the active segment when it is non-empty, expired, and fully committed. */
- private void maybeRollExpiredActiveSegment(long now, LogSegment activeSegment)
+ /** Returns whether the active segment is non-empty, expired, and fully committed. */
+ private boolean shouldRollExpiredActiveSegment(long now, LogSegment activeSegment)
throws IOException {
- if (rollExpiredActiveSegmentEnabled
+ return rollExpiredActiveSegmentEnabled.get()
&& activeSegment.getSizeInBytes() > 0
&& isSegmentExpired(now, activeSegment, logTtlMs)
- && getHighWatermark() >= localLogEndOffset()) {
- roll(Optional.empty());
- }
+ && getHighWatermark() >= localLogEndOffset();
}
@FunctionalInterface
- private interface DeletableSegmentsFinder {
- List<LogSegment> find(long endOffset) throws IOException;
+ private interface SegmentCleanupAction {
+ void cleanup(long endOffset) throws IOException;
}
private boolean isSegmentExpired(long now, LogSegment segment, long expirationTimeMs)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index af073958b4..1f4ea90b9c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -333,6 +333,8 @@ public class TabletServer extends ServerBase {
requestsMetrics);
dynamicConfigManager.register(lakeCatalogDynamicLoader);
+ // Register logManager for dynamic log retention configuration.
+ dynamicConfigManager.register(logManager);
// Register kvManager to dynamicConfigManager for dynamic reconfiguration
dynamicConfigManager.register(kvManager);
// Register DefaultSnapshotContext for dynamic kv.snapshot.interval
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
index 8f51d9d151..5bc4a1fe8c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
@@ -741,6 +741,41 @@ public class DynamicConfigChangeTest {
assertThat(reconfiguredValue.get()).isEqualTo(2);
}
+ @Test
+ void testDynamicRollActiveSegmentChange() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED, false);
+
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
+
+ AtomicReference<Boolean> reconfiguredValue = new AtomicReference<>(false);
+ dynamicConfigManager.register(
+ new ServerReconfigurable() {
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {}
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+ reconfiguredValue.set(
+ newConfig.get(
+ ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED));
+ }
+ });
+ dynamicConfigManager.startup();
+
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED.key(),
+ "true",
+ AlterConfigOpType.SET)));
+
+ Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED.key()))
+ .isEqualTo("true");
+ assertThat(reconfiguredValue.get()).isTrue();
+ }
+
@Test
void testExplicitDataLakeEnabledRequiresDataLakeFormat() throws Exception {
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java
index d5fe4b8b4c..28f9df80f9 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.kv;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.TableConfig;
import org.apache.fluss.memory.TestingMemorySegmentPool;
@@ -60,6 +61,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.fluss.compression.ArrowCompressionInfo.DEFAULT_COMPRESSION;
import static org.apache.fluss.record.LogRecordBatch.CURRENT_LOG_MAGIC_VALUE;
@@ -125,6 +127,8 @@ class KvTabletMergeModeTest {
physicalTablePath,
logTabletDir,
conf,
+ new AtomicBoolean(
+ conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
TestingMetricGroups.TABLET_SERVER_METRICS,
0,
new FlussScheduler(1),
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java
index f3658e9b0f..77539d10b5 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.kv;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.TableConfig;
import org.apache.fluss.memory.TestingMemorySegmentPool;
@@ -54,6 +55,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.fluss.compression.ArrowCompressionInfo.DEFAULT_COMPRESSION;
import static org.apache.fluss.record.LogRecordBatch.CURRENT_LOG_MAGIC_VALUE;
@@ -113,6 +115,8 @@ class KvTabletSchemaEvolutionTest {
physicalTablePath,
logTabletDir,
conf,
+ new AtomicBoolean(
+ conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
TestingMetricGroups.TABLET_SERVER_METRICS,
0,
new FlussScheduler(1),
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
index 38e4be36fd..7b39bd8dac 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -103,6 +103,7 @@ import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -192,6 +193,8 @@ class KvTabletTest {
tablePath,
logTabletDir,
conf,
+ new AtomicBoolean(
+ conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
TestingMetricGroups.TABLET_SERVER_METRICS,
0,
new FlussScheduler(1),
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java
index e9e8ba3240..5e939c9245 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java
@@ -54,6 +54,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.fluss.compression.ArrowCompressionInfo.DEFAULT_COMPRESSION;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK;
@@ -103,6 +104,8 @@ class ScannerManagerTest {
physicalTablePath,
logTabletDir,
conf,
+ new AtomicBoolean(
+ conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
TestingMetricGroups.TABLET_SERVER_METRICS,
0,
new FlussScheduler(1),
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 267b880511..237a3a3418 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
@@ -56,6 +56,7 @@ final class LocalSegmentTTLTest extends ReplicaTestBase {
? 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);
LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
index 075af47c48..88fec0db5c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
@@ -53,6 +53,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
@@ -650,6 +651,8 @@ final class LogLoaderTest extends LogTestBase {
PhysicalTablePath.of(DATA1_TABLE_PATH),
logDir,
conf,
+ new AtomicBoolean(
+ conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
TestingMetricGroups.TABLET_SERVER_METRICS,
recoveryPoint,
scheduler,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
index 3df6a357e9..617257ecb6 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server.log;
import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
@@ -186,6 +187,26 @@ final class LogManagerTest extends LogTestBase {
assertThat(log.isPresent()).isFalse();
}
+ @Test
+ void testReconfigureRollExpiredActiveSegment() throws Exception {
+ initTableBuckets(null);
+ LogTablet existingLog = getOrCreateLog(tablePath1, null, tableBucket1);
+ assertThat(existingLog.isRollExpiredActiveSegmentEnabled()).isFalse();
+
+ Configuration newConfig = new Configuration(conf);
+ newConfig.set(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED, true);
+ logManager.reconfigure(newConfig);
+
+ assertThat(existingLog.isRollExpiredActiveSegmentEnabled()).isTrue();
+ LogTablet newLog = getOrCreateLog(tablePath2, null, tableBucket2);
+ assertThat(newLog.isRollExpiredActiveSegmentEnabled()).isTrue();
+
+ newConfig.set(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED, false);
+ logManager.reconfigure(newConfig);
+ assertThat(existingLog.isRollExpiredActiveSegmentEnabled()).isFalse();
+ assertThat(newLog.isRollExpiredActiveSegmentEnabled()).isFalse();
+ }
+
@ParameterizedTest
@MethodSource("partitionProvider")
void testCheckpointRecoveryPoints(String partitionName) throws Exception {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
index bfb6a730ce..bd634a707c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
@@ -52,6 +52,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.fluss.record.TestData.DATA1;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
@@ -92,6 +93,8 @@ final class LogTabletTest extends LogTestBase {
PhysicalTablePath.of(DATA1_TABLE_PATH),
logDir,
conf,
+ new AtomicBoolean(
+ conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
TestingMetricGroups.TABLET_SERVER_METRICS,
0,
scheduler,
@@ -510,6 +513,8 @@ final class LogTabletTest extends LogTestBase {
PhysicalTablePath.of(DATA1_TABLE_PATH),
logDir,
config,
+ new AtomicBoolean(
+ config.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)),
TestingMetricGroups.TABLET_SERVER_METRICS,
0,
scheduler,
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 f543ee58f5..080eeb5805 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
@@ -56,6 +56,7 @@ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase {
? 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);
LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
@@ -100,6 +101,7 @@ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase {
? 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);
LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
@@ -136,6 +138,7 @@ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase {
? 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);
LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
@@ -156,6 +159,7 @@ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase {
? 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);
LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();