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 875bef312a..4d81d7c64f 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 @@ -1048,7 +1048,7 @@ private void maybeRoll(int messageSize, LogAppendInfo appendInfo) throws Excepti */ @VisibleForTesting @SuppressWarnings("OptionalUsedAsFieldOrParameterType") - public void roll(Optional expectedNextOffset) throws Exception { + public void roll(Optional expectedNextOffset) throws IOException { synchronized (lock) { LogSegment segment = localLog.roll(expectedNextOffset); // Take a snapshot of the writer state to facilitate recovery. It is useful to have @@ -1350,7 +1350,12 @@ private List deletableRemoteSegments(long endOffset) { return deletableSegments; } - /** Returns the contiguous prefix of inactive segments that has expired. */ + /** + * Returns the contiguous prefix of expired inactive segments. + * + *

If all inactive segments are deletable, this also checks whether the active segment should + * be rolled. + */ private List deletableExpiredSegments(long endOffset) throws IOException { if (localLog.getSegments().isEmpty()) { return Collections.emptyList(); @@ -1367,9 +1372,23 @@ private List deletableExpiredSegments(long endOffset) throws IOExcep } 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) + throws IOException { + if (activeSegment.getSizeInBytes() > 0 + && isSegmentExpired(now, activeSegment, logTtlMs) + && getHighWatermark() >= localLogEndOffset()) { + roll(Optional.empty()); + } + } + @FunctionalInterface private interface DeletableSegmentsFinder { List find(long endOffset) throws IOException; 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 new file mode 100644 index 0000000000..427ad90f5e --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LocalSegmentTTLTest.java @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.log; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.server.replica.ReplicaTestBase; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; +import java.util.Collections; + +import static org.apache.fluss.record.TestData.DATA1_SCHEMA; +import static org.apache.fluss.record.TestData.DATA1_TABLE_ID; +import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests TTL-based cleanup when remote log is disabled. */ +final class LocalSegmentTTLTest extends ReplicaTestBase { + + @BeforeEach + public void setup() throws Exception { + super.setup(); + registerTableInZkClient( + DATA1_TABLE_PATH, + DATA1_SCHEMA, + DATA1_TABLE_ID, + Collections.emptyList(), + Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "1h")); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testExpiredActiveSegmentCleaned(boolean partitionTable) throws Exception { + TableBucket tb = + partitionTable + ? new TableBucket(DATA1_TABLE_ID, 0L, 0) + : new TableBucket(DATA1_TABLE_ID, 0); + makeLogTableAsLeader(tb, partitionTable); + LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet(); + + assertThatThrownBy(() -> remoteLogManager.remoteLogTablet(tb)) + .isInstanceOf(IllegalStateException.class); + addMultiSegmentsToLogTablet(logTablet, 1); + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(1); + assertThat(logTablet.localLogStartOffset()).isEqualTo(0L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(0L); + + manualClock.advanceTime(Duration.ofHours(2)); + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(2); + assertThat(logTablet.localLogStartOffset()).isEqualTo(0L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(10L); + + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(1); + assertThat(logTablet.localLogStartOffset()).isEqualTo(10L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(10L); + } +} 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 1ea904ef9a..15328bc6e9 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 @@ -19,6 +19,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.junit.jupiter.api.BeforeEach; @@ -33,7 +34,7 @@ import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH; import static org.assertj.core.api.Assertions.assertThat; -/** Tests TTL-based cleanup of inactive local segments retained by tiered storage. */ +/** Tests TTL-based cleanup of local segments retained by tiered storage. */ final class TieredLocalSegmentTTLTest extends RemoteLogTestBase { @BeforeEach @@ -71,17 +72,28 @@ void testInactiveTieredLocalSegmentRemovedAfterTtl(boolean partitionTable) throw // Run local retention without updating the remote manifest or uploading another segment. logManager.cleanupExpiredLocalLogSegments(); - // The inactive segment is expired and deleted, while the active segment is retained. + // The expired active segment is rolled, but is not deleted in the same retention pass. assertThat(remoteLog.allRemoteLogSegments()).hasSize(4); - assertThat(logTablet.getSegments()).hasSize(1); + assertThat(logTablet.getSegments()).hasSize(2); assertThat(logTablet.localLogStartOffset()).isEqualTo(40L); - assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L); + + remoteLogTaskScheduler.triggerPeriodicScheduledTasks(); + assertThat(remoteLog.getRemoteLogEndOffset()).hasValue(50L); + logManager.cleanupExpiredLocalLogSegments(); + assertThat(logTablet.getSegments()).hasSize(1); + assertThat(logTablet.localLogStartOffset()).isEqualTo(50L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L); + + LogSegment emptyActiveSegment = logTablet.activeLogSegment(); + logManager.cleanupExpiredLocalLogSegments(); + assertThat(logTablet.activeLogSegment()).isSameAs(emptyActiveSegment); } @ParameterizedTest @ValueSource(booleans = {true, false}) - void testExpiredLocalSegmentsRemovedWithoutRemoteLogEndOffset(boolean partitionTable) - throws Exception { + void testExpiredActiveSegmentWaitsForHighWatermarkWithoutRemoteLogEndOffset( + boolean partitionTable) throws Exception { TableBucket tb = partitionTable ? new TableBucket(DATA1_TABLE_ID, 0L, 0) @@ -93,11 +105,44 @@ void testExpiredLocalSegmentsRemovedWithoutRemoteLogEndOffset(boolean partitionT assertThat(remoteLogManager.remoteLogTablet(tb).getRemoteLogEndOffset()).isEmpty(); manualClock.advanceTime(Duration.ofHours(2)); + logTablet.updateHighWatermark(logTablet.localLogEndOffset() - 1L); logManager.cleanupExpiredLocalLogSegments(); assertThat(logTablet.getSegments()).hasSize(1); assertThat(logTablet.localLogStartOffset()).isEqualTo(40L); assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L); + + logTablet.updateHighWatermark(logTablet.localLogEndOffset()); + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(2); + assertThat(logTablet.localLogStartOffset()).isEqualTo(40L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testExpiredActiveSegmentNotRolledWhenTtlDisabled(boolean partitionTable) throws Exception { + registerTableInZkClient( + DATA1_TABLE_PATH, + DATA1_SCHEMA, + DATA1_TABLE_ID, + Collections.emptyList(), + Collections.singletonMap(ConfigOptions.TABLE_LOG_TTL.key(), "0ms")); + TableBucket tb = + partitionTable + ? new TableBucket(DATA1_TABLE_ID, 0L, 0) + : new TableBucket(DATA1_TABLE_ID, 0); + makeLogTableAsLeader(tb, partitionTable); + LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet(); + + addMultiSegmentsToLogTablet(logTablet, 5); + manualClock.advanceTime(Duration.ofHours(2)); + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(5); + assertThat(logTablet.localLogStartOffset()).isEqualTo(0L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L); } @ParameterizedTest @@ -120,5 +165,12 @@ void testTtlCleanupBoundedByRemoteLogEndOffset(boolean partitionTable) throws Ex assertThat(logTablet.getSegments()).hasSize(3); assertThat(logTablet.localLogStartOffset()).isEqualTo(20L); assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(40L); + + logTablet.updateRemoteLogEndOffset(40L); + logManager.cleanupExpiredLocalLogSegments(); + + assertThat(logTablet.getSegments()).hasSize(2); + assertThat(logTablet.localLogStartOffset()).isEqualTo(40L); + assertThat(logTablet.activeLogSegment().getBaseOffset()).isEqualTo(50L); } }