diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java index f7c1b9bf6b4..2fe061f0bca 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java @@ -203,8 +203,7 @@ private CompletedSnapshotStore createCompletedSnapshotStore( retrievedSnapshots.add( checkNotNull(snapshotStateHandle.retrieveCompleteSnapshot())); } catch (Exception e) { - if (e.getMessage() - .contains(CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE)) { + if (CompletedSnapshot.isSnapshotDataNotExists(e)) { LOG.error( "Metadata not found for snapshot {} of table bucket {}, maybe snapshot already removed or broken.", snapshotStateHandle.getSnapshotId(), diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java index 86cd25858b0..830a8890b61 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java @@ -88,6 +88,9 @@ public class CompletedSnapshot { public static final String SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE = "No such file or directory"; + private static final String HADOOP_SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE = + "File does not exist"; + public CompletedSnapshot( TableBucket tableBucket, long snapshotID, @@ -163,9 +166,8 @@ public void registerSharedKvFilesAfterRestored(SharedKvFileRegistry sharedKvFile } public CompletableFuture discardAsync(Executor ioExecutor) { - // it'll discard the snapshot files for kv, it'll always discard - // the private files; for shared files, only if they're not be registered in - // SharedKvRegistry, can the files be deleted. + // Always discard private files. Shared files are discarded directly only when the handle + // represents a newly created snapshot that has not been registered. CompletableFuture discardKvFuture = FutureUtils.runAsync(kvSnapshotHandle::discard, ioExecutor); @@ -195,6 +197,24 @@ public static FsPath getMetadataFilePath(FsPath snapshotLocation) { return new FsPath(snapshotLocation, SNAPSHOT_METADATA_FILE_NAME); } + /** + * Returns whether the throwable or any of its causes indicates that snapshot data no longer + * exists in remote storage. + */ + public static boolean isSnapshotDataNotExists(Throwable throwable) { + Throwable cause = throwable; + while (cause != null) { + String message = cause.getMessage(); + if (message != null + && (message.contains(SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE) + || message.contains(HADOOP_SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE))) { + return true; + } + cause = cause.getCause(); + } + return false; + } + private void disposeMetadata() throws IOException { FsPath metadataFilePath = getMetadataFilePath(); FileSystem fileSystem = metadataFilePath.getFileSystem(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java index 99a7c3d230e..3c9201ff80e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerde.java @@ -193,7 +193,7 @@ public CompletedSnapshot deserialize(JsonNode node) { // construct CompletedSnapshot KvSnapshotHandle kvSnapshotHandle = - new KvSnapshotHandle(sharedFileHandles, privateFileHandles, incrementalSize); + KvSnapshotHandle.restore(sharedFileHandles, privateFileHandles, incrementalSize); Long rowCount = null; if (node.has(ROW_COUNT)) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java index a016f987551..485ac0334cd 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java @@ -27,7 +27,6 @@ import java.util.stream.Collectors; import static org.apache.fluss.utils.Preconditions.checkNotNull; -import static org.apache.fluss.utils.Preconditions.checkState; /** * A handle to the snapshot of a kv tablet. It contains the share file handles and the private file @@ -46,22 +45,36 @@ public class KvSnapshotHandle { private final long incrementalSize; /** - * Once the shared states are registered, it is the {@link SharedKvFileRegistry}'s - * responsibility to cleanup those shared states. But in the cases where the state handle is - * discarded before performing the registration, the handle should delete all the shared states - * created by it. - * - *

This variable is not null iff the handles was registered. + * Whether {@link #discard()} should directly delete the shared files. This is true for newly + * created handles until registration and false for restored handles. */ - private transient SharedKvFileRegistry sharedKvFileRegistry; + private boolean ownsSharedFiles; - public KvSnapshotHandle( + KvSnapshotHandle( List sharedFileHandles, List privateFileHandles, - long incrementalSize) { + long incrementalSize, + boolean ownsSharedFiles) { this.sharedFileHandles = sharedFileHandles; this.privateFileHandles = privateFileHandles; this.incrementalSize = incrementalSize; + this.ownsSharedFiles = ownsSharedFiles; + } + + /** Creates a handle that directly discards its shared files until they are registered. */ + public static KvSnapshotHandle create( + List sharedFileHandles, + List privateFileHandles, + long incrementalSize) { + return new KvSnapshotHandle(sharedFileHandles, privateFileHandles, incrementalSize, true); + } + + /** Restores a handle that only references already-persisted shared files. */ + static KvSnapshotHandle restore( + List sharedFileHandles, + List privateFileHandles, + long incrementalSize) { + return new KvSnapshotHandle(sharedFileHandles, privateFileHandles, incrementalSize, false); } public List getSharedKvFileHandles() { @@ -98,9 +111,6 @@ public long getSnapshotSize() { } public void discard() { - SharedKvFileRegistry registry = this.sharedKvFileRegistry; - final boolean isRegistered = (registry != null); - try { SnapshotUtil.bestEffortDiscardAllKvFiles( privateFileHandles.stream() @@ -110,7 +120,7 @@ public void discard() { LOG.warn("Could not properly discard misc file states.", e); } - if (!isRegistered) { + if (ownsSharedFiles) { try { SnapshotUtil.bestEffortDiscardAllKvFiles( sharedFileHandles.stream() @@ -123,11 +133,8 @@ public void discard() { } public void registerKvFileHandles(SharedKvFileRegistry registry, long snapshotID) { - checkState( - sharedKvFileRegistry != registry, - "The kv file handle has already registered its shared kv files to the given registry."); - - sharedKvFileRegistry = checkNotNull(registry); + checkNotNull(registry); + ownsSharedFiles = false; for (KvFileHandleAndLocalPath handleAndLocalPath : sharedFileHandles) { registry.registerReference( 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 3ae7318f6cb..c50fe3ef06c 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 @@ -221,7 +221,7 @@ public SnapshotResult get(CloseableRegistry snapshotCloseableRegistry) throws Ex completed = true; // We make the 'sstFiles' as the 'shared' in KvSnapshotHandle, final KvSnapshotHandle kvSnapshotHandle = - new KvSnapshotHandle(sstFiles, miscFiles, snapshotIncrementalSize); + KvSnapshotHandle.create(sstFiles, miscFiles, snapshotIncrementalSize); return new SnapshotResult( kvSnapshotHandle, snapshotLocation.getSnapshotDirectory(), tabletState); } finally { 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 f3d1ff76af1..e8fcb24ae9f 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 @@ -854,7 +854,7 @@ private void downloadKvSnapshots(CompletedSnapshot completedSnapshot, Path kvTab try { kvSnapshotDataDownloader.transferAllDataToDirectory(downloadSpec, closeableRegistry); } catch (Exception e) { - if (e.getMessage().contains(CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE)) { + if (CompletedSnapshot.isSnapshotDataNotExists(e)) { try { snapshotContext.handleSnapshotBroken(completedSnapshot); } catch (Exception t) { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java index 2558be96f90..a702af25ccf 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java @@ -186,6 +186,20 @@ void testRemoveCompletedSnapshotStoreFromManager() throws Exception { @Test void testMetadataInconsistencyWithMetadataNotExistsException() throws Exception { + verifyMissingSnapshotMetadataIsCleanedUp( + new FileNotFoundException( + CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE)); + } + + @Test + void testMetadataInconsistencyWithHadoopFileDoesNotExistException() throws Exception { + verifyMissingSnapshotMetadataIsCleanedUp( + new IOException( + "Failed to open snapshot metadata.", + new IOException("File does not exist: /remote/snapshot/_METADATA"))); + } + + private void verifyMissingSnapshotMetadataIsCleanedUp(IOException exception) throws Exception { // setup test data with mixed valid and invalid snapshots TableBucket tableBucket = new TableBucket(1, 1); CompletedSnapshot validSnapshot = @@ -196,7 +210,7 @@ void testMetadataInconsistencyWithMetadataNotExistsException() throws Exception CompletedSnapshot invalidSnapshot = KvTestUtils.mockCompletedSnapshot(tempDir, tableBucket, 2L); TestingCompletedSnapshotHandle invalidSnapshotHandle = - new TestingCompletedSnapshotHandleWithFileNotFound(invalidSnapshot); + new TestingBrokenCompletedSnapshotHandle(invalidSnapshot, exception); // create CompletedSnapshotHandleStore with real implementations CompletedSnapshotHandleStore completedSnapshotHandleStore = @@ -220,6 +234,7 @@ void testMetadataInconsistencyWithMetadataNotExistsException() throws Exception DATA1_TABLE_PATH, tableBucket); assertThat(completedSnapshotStore.getAllSnapshots()).hasSize(1); assertThat(completedSnapshotStore.getAllSnapshots().get(0).getSnapshotID()).isEqualTo(1L); + assertThat(completedSnapshotHandleStore.get(tableBucket, 2L)).isEmpty(); } private CompletedSnapshotStoreManager createCompletedSnapshotStoreManager( @@ -281,17 +296,20 @@ private Set createTableBuckets(int tableNum, int bucketNum) { * A test-specific implementation of CompletedSnapshotHandle that throws FileNotFoundException * with the specific error message expected by CompletedSnapshotStoreManager. */ - private static class TestingCompletedSnapshotHandleWithFileNotFound + private static class TestingBrokenCompletedSnapshotHandle extends TestingCompletedSnapshotHandle { - public TestingCompletedSnapshotHandleWithFileNotFound(CompletedSnapshot snapshot) { + private final IOException exception; + + public TestingBrokenCompletedSnapshotHandle( + CompletedSnapshot snapshot, IOException exception) { super(snapshot, false); + this.exception = exception; } @Override public CompletedSnapshot retrieveCompleteSnapshot() throws IOException { - throw new FileNotFoundException( - CompletedSnapshot.SNAPSHOT_DATA_NOT_EXISTS_ERROR_MESSAGE); + throw exception; } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java index d5bb6301551..00e6ea5e498 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServiceOrphanRpcsITCase.java @@ -221,7 +221,7 @@ private CompletedSnapshot createMinimalSnapshot(TableBucket tb, long snapshotId) tb, snapshotId, new FsPath(snapDir.resolve("_metadata").toUri().toString()), - new KvSnapshotHandle(Collections.emptyList(), Collections.emptyList(), 0L), + KvSnapshotHandle.create(Collections.emptyList(), Collections.emptyList(), 0L), snapshotId, null, null); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java index 2bd93397789..117fef36349 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotJsonSerdeTest.java @@ -56,7 +56,7 @@ protected CompletedSnapshot[] createObjects() { new TableBucket(1, 1), 1, new FsPath("oss://bucket/snapshot"), - new KvSnapshotHandle(sharedFileHandles, privateFileHandles, 5), + KvSnapshotHandle.create(sharedFileHandles, privateFileHandles, 5), 10, null, null); @@ -65,7 +65,7 @@ protected CompletedSnapshot[] createObjects() { new TableBucket(1, 10L, 1), 1, new FsPath("oss://bucket/snapshot"), - new KvSnapshotHandle(sharedFileHandles, privateFileHandles, 5), + KvSnapshotHandle.create(sharedFileHandles, privateFileHandles, 5), 10, 1234L, Collections.singletonList(new AutoIncIDRange(2, 10000, 20000))); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java index 659ab3fbeb4..92faa88392c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java @@ -746,7 +746,7 @@ private CompletedSnapshot getSnapshot(long id) { tableBucket, id, new FsPath(tempDir.toString(), "test_snapshot"), - new KvSnapshotHandle(Collections.emptyList(), Collections.emptyList(), 0)); + KvSnapshotHandle.create(Collections.emptyList(), Collections.emptyList(), 0)); } private CompletedSnapshot getSnapshotWithSharedFiles( @@ -756,7 +756,7 @@ private CompletedSnapshot getSnapshotWithSharedFiles( tableBucket, id, new FsPath(tempDir.toString(), "snapshot_" + id), - new KvSnapshotHandle(sharedFiles, Collections.emptyList(), 0)); + KvSnapshotHandle.create(sharedFiles, Collections.emptyList(), 0)); } private void testSnapshotRetention( @@ -804,7 +804,7 @@ private List> createSnapshotHandles( new TableBucket(1, 1), i, new FsPath("test_snapshot"), - new KvSnapshotHandle( + KvSnapshotHandle.create( Collections.emptyList(), Collections.emptyList(), -1)); final CompletedSnapshotHandle snapshotStateHandle = new TestingCompletedSnapshotHandle( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java index 44b6d4f3b68..323fcb88e98 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotTest.java @@ -27,6 +27,7 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; +import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; import java.util.List; @@ -105,8 +106,32 @@ void testKvSnapshotSize(@TempDir Path tempDir) throws Exception { assertThat(snapshot.getKvSnapshotHandle().getSnapshotSize()).isEqualTo(400L); } + @Test + void testDiscardRestoredSnapshotKeepsSharedFiles(@TempDir Path tempDir) throws Exception { + TableBucket tableBucket = new TableBucket(1, 1); + Path localFileDir = makeDir(tempDir, "local"); + Path snapshotBaseLocation = makeDir(tempDir, "snapshot"); + Path shareDir = makeDir(snapshotBaseLocation, "share"); + Path snapshotPath = makeDir(snapshotBaseLocation, "snapshot-1"); + KvSnapshotHandle kvSnapshotHandle = + makeSnapshotHandle(localFileDir, snapshotPath, shareDir, 100); + CompletedSnapshot snapshot = + new CompletedSnapshot( + tableBucket, + 1, + FsPath.fromLocalFile(snapshotPath.toFile()), + kvSnapshotHandle); + Files.createFile(snapshotPath.resolve("_METADATA")); + + CompletedSnapshot restoredSnapshot = + CompletedSnapshotJsonSerde.fromJson(CompletedSnapshotJsonSerde.toJson(snapshot)); + restoredSnapshot.discardAsync(Executors.directExecutor()).get(); + + checkCompletedSnapshotCleanUp(snapshotPath, restoredSnapshot.getKvSnapshotHandle(), false); + } + private void checkCompletedSnapshotCleanUp( - Path snapshotPath, KvSnapshotHandle kvSnapshotHandle, boolean isShareFileShouldDelete) { + Path snapshotPath, KvSnapshotHandle kvSnapshotHandle, boolean shouldDeleteSharedFiles) { // private should be deleted, but the local file should still remain for (KvFileHandleAndLocalPath kvFileHandleAndLocalPath : kvSnapshotHandle.getPrivateFileHandles()) { @@ -118,8 +143,8 @@ private void checkCompletedSnapshotCleanUp( // check the share files is as expected, and the local file should still remain for (KvFileHandleAndLocalPath kvFileHandleAndLocalPath : kvSnapshotHandle.getSharedKvFileHandles()) { - // share files should also be deleted, but the local file should still remain - if (isShareFileShouldDelete) { + // shared files should also be deleted, but the local file should still remain + if (shouldDeleteSharedFiles) { assertThat(new File(kvFileHandleAndLocalPath.getKvFileHandle().getFilePath())) .doesNotExist(); } else { @@ -168,7 +193,7 @@ KvSnapshotHandle makeSnapshotHandle( new KvFileHandle(privateFile.getPath(), privateFile.length()), localFile.getPath())); } - return new KvSnapshotHandle(sharedFileHandles, privateFileHandles, 10); + return KvSnapshotHandle.create(sharedFileHandles, privateFileHandles, 10); } private Path makeDir(Path basePath, String dirName) { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java index a4b9494a5d8..b97521fca1f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataDownloaderTest.java @@ -162,7 +162,8 @@ private KvSnapshotDownloadSpec createDownloadRequestForContent( handles.get(i), String.format("private-%d-%d", remoteHandleId, i))); } - KvSnapshotHandle kvSnapshotHandle = new KvSnapshotHandle(sharedStates, privateStates, -1); + KvSnapshotHandle kvSnapshotHandle = + KvSnapshotHandle.create(sharedStates, privateStates, -1); return new KvSnapshotDownloadSpec(kvSnapshotHandle, dstPath); } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java index 2e66bd945c9..0ffc45262d7 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/SnapshotsCleanerTest.java @@ -118,7 +118,7 @@ private static class TestKvSnapshotHandle extends KvSnapshotHandle { public TestKvSnapshotHandle( List sharedFileHandles, List privateFileHandles) { - super(sharedFileHandles, privateFileHandles, -1); + super(sharedFileHandles, privateFileHandles, -1, true); } @Override 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 d37e91dc6d5..f3f864e5c4a 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 @@ -19,6 +19,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.exception.OutOfOrderSequenceException; +import org.apache.fluss.fs.FsPath; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.SchemaGetter; @@ -40,6 +41,7 @@ import org.apache.fluss.server.entity.NotifyLeaderAndIsrData; import org.apache.fluss.server.kv.KvTablet; import org.apache.fluss.server.kv.snapshot.CompletedSnapshot; +import org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde; import org.apache.fluss.server.kv.snapshot.TestingCompletedKvSnapshotCommitter; import org.apache.fluss.server.log.FetchParams; import org.apache.fluss.server.log.LogAppendInfo; @@ -50,6 +52,8 @@ import org.apache.fluss.testutils.DataTestUtils; import org.apache.fluss.testutils.common.ManuallyTriggeredScheduledExecutorService; import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.concurrent.Executors; +import org.apache.fluss.utils.function.FunctionWithException; import org.apache.fluss.utils.types.Tuple2; import org.junit.jupiter.api.Test; @@ -64,8 +68,10 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Set; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import static org.apache.fluss.compression.ArrowCompressionInfo.DEFAULT_COMPRESSION; import static org.apache.fluss.record.LogRecordBatch.CURRENT_LOG_MAGIC_VALUE; @@ -601,7 +607,7 @@ void testBrokenSnapshotRecovery(@TempDir File snapshotKvTabletDir) throws Except // trigger second snapshot (may need to wait the task being scheduled) scheduledExecutorService.triggerNextNonPeriodicScheduledTask(Duration.ofSeconds(30)); - kvSnapshotStore.waitUntilSnapshotComplete(tableBucket, 1); + CompletedSnapshot snapshot1 = kvSnapshotStore.waitUntilSnapshotComplete(tableBucket, 1); // put more data and create third snapshot (this will be the broken one) kvRecords = @@ -618,13 +624,24 @@ void testBrokenSnapshotRecovery(@TempDir File snapshotKvTabletDir) throws Except assertThat(kvSnapshotStore.getLatestCompletedSnapshot(tableBucket).getSnapshotID()) .isEqualTo(2); - // now simulate the latest snapshot (snapshot2) being broken by - // deleting its metadata files and unshared SST files - // This simulates file corruption while ZK metadata remains intact - snapshot2.getKvSnapshotHandle().discard(); + Set sharedFilePathsUsedByBothSnapshots = + snapshot1.getKvSnapshotHandle().getSharedKvFileHandles().stream() + .map(handle -> handle.getKvFileHandle().getFilePath()) + .collect(Collectors.toSet()); + sharedFilePathsUsedByBothSnapshots.retainAll( + snapshot2.getKvSnapshotHandle().getSharedKvFileHandles().stream() + .map(handle -> handle.getKvFileHandle().getFilePath()) + .collect(Collectors.toSet())); + assertThat(sharedFilePathsUsedByBothSnapshots).isNotEmpty(); + for (String sharedFilePath : sharedFilePathsUsedByBothSnapshots) { + FsPath path = new FsPath(sharedFilePath); + assertThat(path.getFileSystem().exists(path)).isTrue(); + } - // ZK metadata should still show snapshot2 as latest (file corruption hasn't been detected - // yet) + // Simulate snapshot corruption by deleting only one private file while its metadata and + // shared files remain intact. + assertThat(snapshot2.getKvSnapshotHandle().getPrivateFileHandles()).isNotEmpty(); + snapshot2.getKvSnapshotHandle().getPrivateFileHandles().get(0).getKvFileHandle().discard(); assertThat(kvSnapshotStore.getLatestCompletedSnapshot(tableBucket).getSnapshotID()) .isEqualTo(2); @@ -634,8 +651,31 @@ void testBrokenSnapshotRecovery(@TempDir File snapshotKvTabletDir) throws Except // create a new replica with the same snapshot context // During initialization, it will try to use snapshot2 but find it broken, // then handle the broken snapshot and fall back to snapshot1 + List attemptedSnapshotIds = new ArrayList<>(); testKvSnapshotContext = - new TestSnapshotContext(snapshotKvTabletDir.getPath(), kvSnapshotStore); + new TestSnapshotContext(snapshotKvTabletDir.getPath(), kvSnapshotStore) { + @Override + public FunctionWithException + getLatestCompletedSnapshotProvider() { + return bucket -> { + CompletedSnapshot snapshot = + testKvSnapshotStore.getLatestCompletedSnapshot(bucket); + if (snapshot != null) { + attemptedSnapshotIds.add(snapshot.getSnapshotID()); + return CompletedSnapshotJsonSerde.fromJson( + CompletedSnapshotJsonSerde.toJson(snapshot)); + } + return null; + }; + } + + @Override + public void handleSnapshotBroken(CompletedSnapshot snapshot) throws Exception { + testKvSnapshotStore.removeSnapshot( + snapshot.getTableBucket(), snapshot.getSnapshotID()); + snapshot.discardAsync(Executors.directExecutor()).get(); + } + }; kvReplica = makeKvReplica(DATA1_PHYSICAL_TABLE_PATH_PK, tableBucket, testKvSnapshotContext); // make it leader again - this should trigger the broken snapshot recovery logic @@ -647,19 +687,23 @@ void testBrokenSnapshotRecovery(@TempDir File snapshotKvTabletDir) throws Except assertThat(kvReplica.getKvTablet()).isNotNull(); KvTablet kvTablet = kvReplica.getKvTablet(); - // verify that the data from snapshot1 is restored (snapshot2 was broken and cleaned up) - // snapshot1 should contain: k1->3,c and k3->4,d + assertThat(attemptedSnapshotIds).containsExactly(2L, 1L); + assertThat(kvSnapshotStore.getLatestCompletedSnapshot(tableBucket).getSnapshotID()) + .isEqualTo(1); + for (String sharedFilePath : sharedFilePathsUsedByBothSnapshots) { + FsPath path = new FsPath(sharedFilePath); + assertThat(path.getFileSystem().exists(path)).isTrue(); + } + + // Snapshot1 is restored after snapshot2 is discarded. List> expectedKeyValues = getKeyValuePairs( genKvRecords( Tuple2.of("k1", new Object[] {3, "c"}), + Tuple2.of("k2", new Object[] {2, "b"}), Tuple2.of("k3", new Object[] {4, "d"}))); verifyGetKeyValues(kvTablet, expectedKeyValues); - // Verify the core functionality: KvTablet successfully initialized despite broken snapshot - // The key test is that the system can handle broken snapshots and recover correctly - - // Verify that we successfully simulated the broken snapshot condition File metadataFile = new File(snapshot2.getMetadataFilePath().getPath()); assertThat(metadataFile.exists()).isFalse(); } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java index 1ff32c7e7a9..0e9de09dd76 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/KvTestUtils.java @@ -128,7 +128,7 @@ public static CompletedSnapshot mockCompletedSnapshot( tableBucket.getTableId(), tableBucket.getBucket(), snapshotId)), - new KvSnapshotHandle(Collections.emptyList(), Collections.emptyList(), 0), + KvSnapshotHandle.create(Collections.emptyList(), Collections.emptyList(), 0), 0, null, null);