diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java index c1ee5f86761..273b8978c58 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java @@ -40,6 +40,8 @@ import org.apache.fluss.server.kv.autoinc.AutoIncrementManager; import org.apache.fluss.server.kv.autoinc.ZkSequenceGeneratorFactory; import org.apache.fluss.server.kv.rowmerger.RowMerger; +import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.server.kv.snapshot.LocalKvSnapshotUtils; import org.apache.fluss.server.log.LogManager; import org.apache.fluss.server.log.LogTablet; import org.apache.fluss.server.metrics.group.TabletServerMetricGroup; @@ -341,7 +343,7 @@ public KvTablet getOrCreateKv( * @param tableBucket the table bucket * @return the tablet directory */ - public File createTabletDir( + public File deleteAndCreateTabletDir( File dataDir, PhysicalTablePath tablePath, TableBucket tableBucket) { File tabletDir = getTabletDir(dataDir, tablePath, tableBucket); @@ -351,6 +353,34 @@ public File createTabletDir( return tabletDir; } + /** + * Attempts to rebuild a tablet directory from the matching retained local snapshot. + * + * @return the rebuilt tablet directory, or empty when the local snapshot is unavailable or + * invalid + */ + public Optional restoreKvFromLocalSnapshot( + File dataDir, + PhysicalTablePath tablePath, + TableBucket tableBucket, + CompletedSnapshot completedSnapshot) { + if (!tableBucket.equals(completedSnapshot.getTableBucket())) { + return Optional.empty(); + } + + File tabletDir = getTabletDir(dataDir, tablePath, tableBucket); + if (!tabletDir.isDirectory()) { + LOG.debug( + "Skip retained local snapshot recovery because KV tablet directory {} " + + "does not exist or is not a directory.", + tabletDir); + return Optional.empty(); + } + return LocalKvSnapshotUtils.restore(tabletDir, completedSnapshot) + ? Optional.of(tabletDir) + : Optional.empty(); + } + public Optional getKv(TableBucket tableBucket) { return Optional.ofNullable(currentKvs.get(tableBucket)); } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/LocalKvSnapshotUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/LocalKvSnapshotUtils.java new file mode 100644 index 00000000000..cf141b7c397 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/LocalKvSnapshotUtils.java @@ -0,0 +1,208 @@ +/* + * 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.kv.snapshot; + +import org.apache.fluss.server.kv.rocksdb.RocksDBKvBuilder; +import org.apache.fluss.utils.FileUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.HashMap; +import java.util.Map; + +/** Utilities for retaining, validating, and restoring local KV snapshot checkpoints. */ +public final class LocalKvSnapshotUtils { + + private static final Logger LOG = LoggerFactory.getLogger(LocalKvSnapshotUtils.class); + + private static final String SNAPSHOT_DIRECTORY_PREFIX = "snap-"; + private static final String RESTORE_DIRECTORY_PREFIX = ".db-restore-"; + + private LocalKvSnapshotUtils() {} + + /** Returns the local checkpoint directory for the given snapshot. */ + public static File getSnapshotDirectory(File kvTabletDir, long snapshotId) { + return new File(kvTabletDir, SNAPSHOT_DIRECTORY_PREFIX + snapshotId); + } + + /** + * Rebuilds the active RocksDB directory from a matching local checkpoint. + * + *

The checkpoint is accepted only when every file referenced by the committed remote + * snapshot metadata exists locally with the expected size and there are no unexpected files. + * SST files are hard-linked into a temporary RocksDB directory when possible; mutable metadata + * files are copied. The retained checkpoint itself is left untouched so another local restart + * can use it before a newer snapshot completes. + * + * @return whether the active RocksDB directory was rebuilt from the local checkpoint + */ + public static boolean restore(File kvTabletDir, CompletedSnapshot completedSnapshot) { + long snapshotId = completedSnapshot.getSnapshotID(); + File snapshotDirectory = getSnapshotDirectory(kvTabletDir, snapshotId); + Map expectedFiles = getExpectedFiles(completedSnapshot); + if (!isValid(snapshotDirectory.toPath(), expectedFiles)) { + if (snapshotDirectory.exists()) { + LOG.warn( + "Retained local KV snapshot {} does not match committed snapshot metadata. " + + "Falling back to remote snapshot recovery.", + snapshotDirectory); + } + return false; + } + + Path restoreDirectory = kvTabletDir.toPath().resolve(RESTORE_DIRECTORY_PREFIX + snapshotId); + Path activeDbDirectory = RocksDBKvBuilder.getInstanceRocksDBPath(kvTabletDir).toPath(); + try { + FileUtils.deleteDirectory(restoreDirectory.toFile()); + Files.createDirectories(restoreDirectory); + for (String fileName : expectedFiles.keySet()) { + Path source = snapshotDirectory.toPath().resolve(fileName); + Path target = restoreDirectory.resolve(fileName); + copySnapshotFile(source, target); + } + + FileUtils.deleteDirectory(activeDbDirectory.toFile()); + FileUtils.atomicMoveWithFallback(restoreDirectory, activeDbDirectory); + retainOnly(kvTabletDir, snapshotId); + return true; + } catch (Exception e) { + LOG.warn( + "Failed to rebuild local KV directory {} from snapshot {}. " + + "Falling back to remote snapshot recovery.", + activeDbDirectory, + snapshotId, + e); + FileUtils.deleteDirectoryQuietly(restoreDirectory.toFile()); + return false; + } + } + + static void retainOnly(File kvTabletDir, long snapshotId) { + File retainedSnapshot = getSnapshotDirectory(kvTabletDir, snapshotId); + for (File snapshotDirectory : FileUtils.listDirectories(kvTabletDir)) { + if (isSnapshotDirectory(snapshotDirectory) + && !snapshotDirectory.equals(retainedSnapshot)) { + deleteQuietly(snapshotDirectory); + } + } + } + + static void discard(File kvTabletDir, long snapshotId) { + deleteQuietly(getSnapshotDirectory(kvTabletDir, snapshotId)); + } + + private static Map getExpectedFiles(CompletedSnapshot completedSnapshot) { + Map expectedFiles = new HashMap<>(); + KvSnapshotHandle snapshotHandle = completedSnapshot.getKvSnapshotHandle(); + for (KvFileHandleAndLocalPath file : snapshotHandle.getSharedKvFileHandles()) { + addExpectedFile(expectedFiles, file); + } + for (KvFileHandleAndLocalPath file : snapshotHandle.getPrivateFileHandles()) { + addExpectedFile(expectedFiles, file); + } + return expectedFiles; + } + + private static void addExpectedFile( + Map expectedFiles, KvFileHandleAndLocalPath file) { + String localPath = file.getLocalPath(); + Long previous = expectedFiles.put(localPath, file.getKvFileHandle().getSize()); + if (previous != null) { + // Duplicate paths cannot describe an unambiguous local checkpoint. + expectedFiles.put(localPath, -1L); + } + } + + private static boolean isValid(Path snapshotDirectory, Map expectedFiles) { + if (!Files.isDirectory(snapshotDirectory, LinkOption.NOFOLLOW_LINKS) + || expectedFiles.isEmpty()) { + return false; + } + + try { + Path[] actualFiles = FileUtils.listDirectory(snapshotDirectory); + if (actualFiles.length != expectedFiles.size()) { + return false; + } + + for (Map.Entry expectedFile : expectedFiles.entrySet()) { + Path relativePath = Paths.get(expectedFile.getKey()).normalize(); + if (relativePath.isAbsolute() + || relativePath.getNameCount() != 1 + || !expectedFile.getKey().equals(relativePath.toString()) + || expectedFile.getValue() < 0) { + return false; + } + + Path localFile = snapshotDirectory.resolve(relativePath).normalize(); + if (!localFile.getParent().equals(snapshotDirectory.normalize()) + || !Files.isRegularFile(localFile, LinkOption.NOFOLLOW_LINKS) + || Files.size(localFile) != expectedFile.getValue()) { + return false; + } + } + + for (Path actualFile : actualFiles) { + if (!expectedFiles.containsKey(actualFile.getFileName().toString())) { + return false; + } + } + return true; + } catch (Exception e) { + LOG.warn("Failed to validate local KV snapshot directory {}.", snapshotDirectory, e); + return false; + } + } + + private static void copySnapshotFile(Path source, Path target) throws IOException { + if (source.getFileName().toString().endsWith(RocksIncrementalSnapshot.SST_FILE_SUFFIX)) { + try { + Files.createLink(target, source); + return; + } catch (UnsupportedOperationException | IOException linkException) { + try { + Files.copy(source, target); + return; + } catch (IOException copyException) { + copyException.addSuppressed(linkException); + throw copyException; + } + } + } + Files.copy(source, target); + } + + private static boolean isSnapshotDirectory(File directory) { + return directory.getName().startsWith(SNAPSHOT_DIRECTORY_PREFIX); + } + + private static void deleteQuietly(File snapshotDirectory) { + try { + FileUtils.deleteDirectory(snapshotDirectory); + } catch (IOException e) { + LOG.warn("Could not properly clean local KV snapshot {}.", snapshotDirectory, e); + } + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java index c50fe3ef06c..4c56b11ec0b 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshot.java @@ -113,12 +113,16 @@ public void notifySnapshotComplete(long completedSnapshotId) { uploadedSstFiles.keySet().removeIf(snapshotId -> snapshotId < completedSnapshotId); lastCompletedSnapshotId = completedSnapshotId; } + // The local checkpoint is useful after an in-place restart. Keep the committed snapshot and + // remove older or uncommitted checkpoints only after the remote commit succeeds. + LocalKvSnapshotUtils.retainOnly(instanceBasePath, completedSnapshotId); } public void notifySnapshotAbort(long abortedSnapshotId) { synchronized (uploadedSstFiles) { uploadedSstFiles.remove(abortedSnapshotId); } + LocalKvSnapshotUtils.discard(instanceBasePath, abortedSnapshotId); } @Override @@ -138,7 +142,7 @@ public NativeRocksDBSnapshotResources syncPrepareResources(long snapshotId) thro } private File prepareLocalSnapshotDirectory(long snapshotId) { - return new File(instanceBasePath, "snap-" + snapshotId); + return LocalKvSnapshotUtils.getSnapshotDirectory(instanceBasePath, snapshotId); } private PreviousSnapshot getPreviousSnapshot(long snapshotId) { @@ -323,16 +327,9 @@ protected NativeRocksDBSnapshotResources( @Override public void release() { - try { - if (snapshotDirectory.exists()) { - LOG.trace( - "Running cleanup for local RocksDB backup directory {}.", - snapshotDirectory); - FileUtils.deleteDirectory(snapshotDirectory); - } - } catch (IOException e) { - LOG.warn("Could not properly cleanup local RocksDB backup directory.", e); - } + // Do not delete the checkpoint when the asynchronous upload finishes. Its lifecycle is + // decided only after the snapshot commit result is known: notifySnapshotComplete keeps + // the latest committed checkpoint and notifySnapshotAbort removes a failed candidate. } } 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 784c4ad047c..9a03f7a530c 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 @@ -787,21 +787,9 @@ private Optional initKvTablet() { long startTime = clock.milliseconds(); LOG.info("Start to init kv tablet for {} of table {}.", tableBucket, physicalPath); - // todo: we may need to handle the following cases: - // case1: no kv files in local, restore from remote snapshot; and apply - // the log; - // case2: kv files in local - // - if no remote snapshot, restore from local and apply the log known to the local - // files. - // - have snapshot, if the known offset to the local files is much less than(maybe - // some value configured) - // the remote snapshot; restore from remote snapshot; - - // currently for simplicity, we'll always download the snapshot files and restore from - // the snapshots as kv files won't exist in our current implementation for - // when replica become follower, we'll always delete the kv files. - - // get the offset from which, we should restore from. default is 0 + // Prefer a retained local checkpoint only when it exactly matches the latest committed + // snapshot metadata. Otherwise, download the committed snapshot from remote storage. + // The default recovery offset is 0 when no committed snapshot exists. long restoreStartOffset = 0; Optional optCompletedSnapshot = getLatestSnapshot(tableBucket); try { @@ -814,12 +802,29 @@ private Optional initKvTablet() { tableBucket, physicalPath); CompletedSnapshot completedSnapshot = optCompletedSnapshot.get(); - // always create a new dir for the kv tablet - File tabletDir = - kvManager.createTabletDir( - logTablet.getDataDir(), physicalPath, tableBucket); - // down the snapshot to target tablet dir - downloadKvSnapshots(completedSnapshot, tabletDir.toPath()); + + File tabletDir; + long start = System.currentTimeMillis(); + Optional optionalTabletDir = + kvManager.restoreKvFromLocalSnapshot( + logTablet.getDataDir(), + physicalPath, + tableBucket, + completedSnapshot); + if (optionalTabletDir.isPresent()) { + tabletDir = optionalTabletDir.get(); + LOG.info( + "Rebuilt kv tablet for {} of table {} from retained local snapshot {} that costs {} ms.", + tableBucket, + physicalPath, + completedSnapshot.getSnapshotID(), + System.currentTimeMillis() - start); + } else { + tabletDir = + kvManager.deleteAndCreateTabletDir( + logTablet.getDataDir(), physicalPath, tableBucket); + downloadKvSnapshots(completedSnapshot, tabletDir.toPath()); + } // as we have downloaded kv files into the tablet dir, now, we can load it kvTablet = kvManager.loadKv(tabletDir, schemaGetter, this::onKvFlushComplete); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/LocalKvSnapshotUtilsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/LocalKvSnapshotUtilsTest.java new file mode 100644 index 00000000000..aef9b52d3cc --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/LocalKvSnapshotUtilsTest.java @@ -0,0 +1,130 @@ +/* + * 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.kv.snapshot; + +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.server.kv.rocksdb.RocksDBKvBuilder; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link LocalKvSnapshotUtils}. */ +class LocalKvSnapshotUtilsTest { + + @Test + void testRestoreMissingTabletDirectory(@TempDir Path dataDir) { + Path missingTabletDir = dataDir.resolve("missing-tablet"); + + assertThat( + LocalKvSnapshotUtils.restore( + missingTabletDir.toFile(), completedSnapshot(1L, 1L, 1L))) + .isFalse(); + assertThat(missingTabletDir).doesNotExist(); + } + + @Test + void testRestoreValidSnapshotAndRetainCheckpoint(@TempDir Path tabletDir) throws Exception { + long snapshotId = 2L; + Path staleSnapshot = + LocalKvSnapshotUtils.getSnapshotDirectory(tabletDir.toFile(), 1L).toPath(); + Files.createDirectories(staleSnapshot); + Files.write(staleSnapshot.resolve("stale"), new byte[] {1}); + + Path snapshotDirectory = + LocalKvSnapshotUtils.getSnapshotDirectory(tabletDir.toFile(), snapshotId).toPath(); + Files.createDirectories(snapshotDirectory); + byte[] sstBytes = "sst-data".getBytes(StandardCharsets.UTF_8); + byte[] currentBytes = "MANIFEST-1".getBytes(StandardCharsets.UTF_8); + Files.write(snapshotDirectory.resolve("000001.sst"), sstBytes); + Files.write(snapshotDirectory.resolve("CURRENT"), currentBytes); + + Path activeDb = RocksDBKvBuilder.getInstanceRocksDBPath(tabletDir.toFile()).toPath(); + Files.createDirectories(activeDb); + Files.write(activeDb.resolve("old"), new byte[] {1}); + + CompletedSnapshot completedSnapshot = + completedSnapshot(snapshotId, sstBytes.length, currentBytes.length); + assertThat(LocalKvSnapshotUtils.restore(tabletDir.toFile(), completedSnapshot)).isTrue(); + + assertThat(activeDb.resolve("old")).doesNotExist(); + assertThat(activeDb.resolve("000001.sst")).hasBinaryContent(sstBytes); + assertThat(activeDb.resolve("CURRENT")).hasBinaryContent(currentBytes); + assertThat(snapshotDirectory).isDirectory(); + assertThat(staleSnapshot).doesNotExist(); + + // Mutable RocksDB metadata must be copied rather than linked back into the checkpoint. + Files.write(activeDb.resolve("CURRENT"), "MANIFEST-2".getBytes(StandardCharsets.UTF_8)); + assertThat(snapshotDirectory.resolve("CURRENT")).hasBinaryContent(currentBytes); + } + + @Test + void testInvalidSnapshotFallsBackWithoutChangingActiveDb(@TempDir Path tabletDir) + throws Exception { + long snapshotId = 3L; + Path snapshotDirectory = + LocalKvSnapshotUtils.getSnapshotDirectory(tabletDir.toFile(), snapshotId).toPath(); + Files.createDirectories(snapshotDirectory); + Files.write( + snapshotDirectory.resolve("000001.sst"), + "wrong-size".getBytes(StandardCharsets.UTF_8)); + Files.write(snapshotDirectory.resolve("CURRENT"), new byte[] {1}); + + Path activeDb = RocksDBKvBuilder.getInstanceRocksDBPath(tabletDir.toFile()).toPath(); + Files.createDirectories(activeDb); + Path activeMarker = activeDb.resolve("active"); + Files.write(activeMarker, new byte[] {1}); + + assertThat( + LocalKvSnapshotUtils.restore( + tabletDir.toFile(), completedSnapshot(snapshotId, 1, 1))) + .isFalse(); + assertThat(activeMarker).exists(); + assertThat(snapshotDirectory).isDirectory(); + } + + private static CompletedSnapshot completedSnapshot( + long snapshotId, long sstSize, long currentSize) { + KvSnapshotHandle snapshotHandle = + KvSnapshotHandle.create( + Collections.singletonList( + KvFileHandleAndLocalPath.of( + new KvFileHandle( + "file:///remote/shared/000001.sst", sstSize), + "000001.sst")), + Collections.singletonList( + KvFileHandleAndLocalPath.of( + new KvFileHandle( + "file:///remote/snapshot/CURRENT", currentSize), + "CURRENT")), + sstSize + currentSize); + return new CompletedSnapshot( + new TableBucket(1L, 0), + snapshotId, + new FsPath(new File("remote-snapshot").toURI()), + snapshotHandle); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java index 64398516abc..348f26acb7b 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/RocksIncrementalSnapshotTest.java @@ -83,12 +83,32 @@ void testIncrementalSnapshot(@TempDir Path snapshotBaseDir, @TempDir Path snapsh // make and notify snapshot with id 1 KvSnapshotHandle kvSnapshotHandle1 = snapshot(1L, incrementalSnapshot, snapshotLocation, closeableRegistry); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 1L)) + .isDirectory(); incrementalSnapshot.notifySnapshotComplete(1L); // make and notify snapshot with id 2 KvSnapshotHandle kvSnapshotHandle2 = snapshot(2L, incrementalSnapshot, snapshotLocation, closeableRegistry); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 1L)) + .isDirectory(); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 2L)) + .isDirectory(); incrementalSnapshot.notifySnapshotComplete(2L); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 1L)) + .doesNotExist(); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 2L)) + .isDirectory(); // the share kv file handles for cp2 should be equal to the handles for cp1 verifyShareFileEqual(kvSnapshotHandle2, kvSnapshotHandle1); // all file handles should be PlaceHolderHandle @@ -103,6 +123,14 @@ void testIncrementalSnapshot(@TempDir Path snapshotBaseDir, @TempDir Path snapsh snapshot(3L, incrementalSnapshot, snapshotLocation, closeableRegistry); // assume it's fail incrementalSnapshot.notifySnapshotAbort(3L); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 3L)) + .doesNotExist(); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 2L)) + .isDirectory(); // write some data again rocksDB.put("key3".getBytes(), "val3".getBytes()); @@ -111,6 +139,15 @@ void testIncrementalSnapshot(@TempDir Path snapshotBaseDir, @TempDir Path snapsh // make sure the uploaded files contains the files in snapshot 3 and snapshot 4 // there're two newly uploaded files, one for cp3, one for cp4 checkSnapshotIncrementWithNewlyFiles(kvSnapshotHandle4, kvSnapshotHandle1, 2); + incrementalSnapshot.notifySnapshotComplete(4L); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 2L)) + .doesNotExist(); + assertThat( + LocalKvSnapshotUtils.getSnapshotDirectory( + rocksDBExtension.getRockDbDir(), 4L)) + .isDirectory(); // now, let try to rebuild from cp2 and cp4 // test restore from cp2 @@ -136,6 +173,7 @@ void testIncrementalSnapshot(@TempDir Path snapshotBaseDir, @TempDir Path snapsh snapshot(5L, incrementalSnapshot, snapshotLocation, closeableRegistry); // discard the snapshot handle kvSnapshotHandle5.discard(); + incrementalSnapshot.notifySnapshotAbort(5L); // we can still restore from cp4 Path dest3 = snapshotDownDir.resolve("restore3"); @@ -189,13 +227,19 @@ public KvSnapshotHandle snapshot( RocksIncrementalSnapshot.NativeRocksDBSnapshotResources nativeRocksDBSnapshotResources = incrementalSnapshot.syncPrepareResources(snapshotId); - return incrementalSnapshot - .asyncSnapshot( - nativeRocksDBSnapshotResources, - snapshotId, - new TabletState(0L, null, null), - snapshotLocation) - .get(closeableRegistry) - .getKvSnapshotHandle(); + try { + return incrementalSnapshot + .asyncSnapshot( + nativeRocksDBSnapshotResources, + snapshotId, + new TabletState(0L, null, null), + snapshotLocation) + .get(closeableRegistry) + .getKvSnapshotHandle(); + } finally { + // Upload completion releases native resources, but the local checkpoint remains until + // the commit result is reported. + nativeRocksDBSnapshotResources.release(); + } } } 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 2d8b888308b..637814bab8b 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 @@ -41,15 +41,18 @@ import org.apache.fluss.rpc.protocol.MergeMode; import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; import org.apache.fluss.server.kv.KvFlushScheduler; +import org.apache.fluss.server.kv.KvManager; import org.apache.fluss.server.kv.KvTablet; import org.apache.fluss.server.kv.TestingHoldableKvFlushScheduler; import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; import org.apache.fluss.server.kv.snapshot.KvSnapshotDataDownloader; import org.apache.fluss.server.kv.snapshot.KvSnapshotDownloadSpec; +import org.apache.fluss.server.kv.snapshot.LocalKvSnapshotUtils; import org.apache.fluss.server.kv.snapshot.TestingCompletedKvSnapshotCommitter; import org.apache.fluss.server.log.FetchParams; import org.apache.fluss.server.log.LogAppendInfo; import org.apache.fluss.server.log.LogReadInfo; +import org.apache.fluss.server.metrics.group.TestingMetricGroups; import org.apache.fluss.server.testutils.KvTestUtils; import org.apache.fluss.server.zk.data.LeaderAndIsr; import org.apache.fluss.testutils.DataTestUtils; @@ -630,22 +633,42 @@ void testKvReplicaSnapshot(@TempDir File snapshotKvTabletDir) throws Exception { completedSnapshot1.getKvSnapshotHandle(), completedSnapshot0.getKvSnapshotHandle(), 1); - // now, make the replica as follower to make kv can be destroyed - makeKvReplicaAsFollower(kvReplica, 1); + File localKvTabletDir = checkNotNull(kvReplica.getKvTablet()).getKvTabletDir(); + File retainedLocalSnapshot = + LocalKvSnapshotUtils.getSnapshotDirectory( + localKvTabletDir, completedSnapshot1.getSnapshotID()); + + // Restart the KV manager without a role transition. A normal shutdown preserves the tablet + // directory, including the retained snapshot checkpoint. + restartKvManager(); + assertThat(retainedLocalSnapshot).isDirectory(); // make a new kv replica + AtomicBoolean downloadedRemoteSnapshot = new AtomicBoolean(false); testKvSnapshotContext = - new TestSnapshotContext(snapshotKvTabletDir.getPath(), kvSnapshotStore); + new TestSnapshotContext(snapshotKvTabletDir.getPath(), kvSnapshotStore) { + @Override + public KvSnapshotDataDownloader getSnapshotDataDownloader() { + return new KvSnapshotDataDownloader(executorService) { + @Override + public void transferAllDataToDirectory( + KvSnapshotDownloadSpec downloadSpec, + CloseableRegistry closeableRegistry) + throws Exception { + downloadedRemoteSnapshot.set(true); + super.transferAllDataToDirectory(downloadSpec, closeableRegistry); + } + }; + } + }; kvReplica = makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, tableBucket, testKvSnapshotContext); scheduledExecutorService = testKvSnapshotContext.scheduledExecutorService; kvSnapshotStore = testKvSnapshotContext.testKvSnapshotStore; - makeKvReplicaAsFollower(kvReplica, 1); - - // check the kv tablet should be null since it has become follower - assertThat(kvReplica.getKvTablet()).isNull(); - // make as leader again, should restore from snapshot + // Recover as leader after the in-place restart, without an intervening role transition. makeKvReplicaAsLeader(kvReplica, 2); + assertThat(downloadedRemoteSnapshot).isFalse(); + assertThat(retainedLocalSnapshot).isDirectory(); // put some data kvRecords = @@ -666,6 +689,14 @@ void testKvReplicaSnapshot(@TempDir File snapshotKvTabletDir) throws Exception { Tuple2.of("k2", new Object[] {4, "bk21"}), Tuple2.of("k3", new Object[] {5, "k3"}))); KvTestUtils.checkSnapshot(completedSnapshot2, expectedKeyValues, expectedLogOffset); + + File finalKvTabletDir = checkNotNull(kvReplica.getKvTablet()).getKvTabletDir(); + File finalLocalSnapshot = + LocalKvSnapshotUtils.getSnapshotDirectory( + finalKvTabletDir, completedSnapshot2.getSnapshotID()); + assertThat(finalLocalSnapshot).isDirectory(); + makeKvReplicaAsFollower(kvReplica, 3); + assertThat(finalKvTabletDir).doesNotExist(); } @Test @@ -1112,6 +1143,18 @@ private Gauge getBucketLocalLogSizeGauge(Replica replica) { .get(MetricNames.BUCKET_PHYSICAL_STORAGE_LOCAL_LOG_SIZE); } + private void restartKvManager() throws IOException { + kvManager.shutdown(); + kvManager = + KvManager.create( + conf, + zkClient, + logManager, + TestingMetricGroups.TABLET_SERVER_METRICS, + localDiskManager); + kvManager.startup(); + } + private void makeLogReplicaAsLeader(Replica replica) throws Exception { makeLeaderReplica( replica,