From f491141520b54f6eec3bde7d97d1368756ecae46 Mon Sep 17 00:00:00 2001 From: yunhong <337361684@qq.com> Date: Fri, 18 Sep 2026 10:27:48 +0800 Subject: [PATCH] [hotfix][server] Simplify KV cleanup tests and directory discovery Consolidate overlapping KV cleanup tests while retaining controlled shutdown, snapshot recovery, and failed-deletion isolation coverage. Use shared directory discovery for logs and KV. Skip symbolic links without following their targets and continue past unreadable directories. Validation: 55 targeted tests passed. Both tablet discovery cases passed again after the final NOFOLLOW_LINKS simplification. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 90/90 AI-Contributed/UT: 274/274 --- .../fluss/server/TabletManagerBase.java | 53 +++--- .../org/apache/fluss/server/kv/KvManager.java | 37 +--- .../fluss/server/TabletManagerBaseTest.java | 37 +--- .../apache/fluss/server/kv/KvManagerTest.java | 161 ++---------------- .../tablet/TabletServerShutdownITCase.java | 76 ++------- 5 files changed, 65 insertions(+), 299 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java b/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java index 187f2c5a18..f462bce331 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java @@ -30,7 +30,6 @@ import org.apache.fluss.server.log.LogManager; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.TableRegistration; -import org.apache.fluss.utils.FileUtils; import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; @@ -40,9 +39,9 @@ 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.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import java.util.LinkedHashMap; import java.util.List; @@ -55,7 +54,6 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; import java.util.function.Predicate; -import java.util.stream.Collectors; import static org.apache.fluss.utils.FlussPaths.HISTORICAL_LOOKUP_CACHE_DIR_NAME; import static org.apache.fluss.utils.FlussPaths.KV_TABLET_DIR_PREFIX; @@ -112,37 +110,26 @@ protected Map> listTabletsToLoad() { return tabletsToLoadByDataDir; } - /** Returns the tablet directories to be loaded from a single configured data directory. */ - protected List listTabletsToLoad(File dataDir) { - return listTabletsToLoad( - dataDir, - (directory, nameFilter) -> - Arrays.stream(FileUtils.listDirectories(directory)) - .filter(file -> nameFilter.test(file.getName())) - .collect(Collectors.toList())); - } - /** - * Lists tablet directories using the common layout and cache exclusions, with a caller-supplied - * directory lister to control symbolic-link handling and listing failures. + * Returns the tablet directories from a single configured data directory, skipping caches, + * symbolic links, and unreadable directories. */ - protected List listTabletsToLoad( - File dataDir, DirectoryLister directoryLister) throws E { + protected List listTabletsToLoad(File dataDir) { List tabletsToLoad = new ArrayList<>(); for (File dbDir : - directoryLister.listDirectories( + listDirectories( dataDir, name -> !name.equals(HISTORICAL_LOOKUP_CACHE_DIR_NAME) && !name.equals(REMOTE_LOG_INDEX_LOCAL_CACHE))) { - for (File tableDir : directoryLister.listDirectories(dbDir, name -> true)) { + for (File tableDir : listDirectories(dbDir, name -> true)) { for (File tabletOrPartitionDir : - directoryLister.listDirectories( + listDirectories( tableDir, name -> isPartitionDir(name) || name.startsWith(tabletDirPrefix))) { if (isPartitionDir(tabletOrPartitionDir.getName())) { tabletsToLoad.addAll( - directoryLister.listDirectories( + listDirectories( tabletOrPartitionDir, name -> name.startsWith(tabletDirPrefix))); } else { @@ -154,14 +141,22 @@ protected List listTabletsToLoad( return tabletsToLoad; } - /** - * Lists child directories whose names match a filter, optionally reporting listing failures. - */ - @FunctionalInterface - protected interface DirectoryLister { - - /** Returns the matching child directories. */ - List listDirectories(File parent, Predicate nameFilter) throws E; + private List listDirectories(File parent, Predicate nameFilter) { + List directories = new ArrayList<>(); + File[] entries = parent.listFiles(); + if (entries == null) { + LOG.warn("Failed to list tablet directories in {}. Skipping this directory.", parent); + return directories; + } + for (File entry : entries) { + if (!nameFilter.test(entry.getName())) { + continue; + } + if (Files.isDirectory(entry.toPath(), LinkOption.NOFOLLOW_LINKS)) { + directories.add(entry); + } + } + return directories; } protected ExecutorService createThreadPool(String poolName) { 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 1ac64e786a..848b9e7288 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 @@ -67,14 +67,10 @@ import java.io.File; import java.io.IOException; -import java.nio.file.DirectoryIteratorException; import java.nio.file.DirectoryNotEmptyException; -import java.nio.file.DirectoryStream; import java.nio.file.Files; -import java.nio.file.LinkOption; import java.nio.file.Path; import java.nio.file.StandardCopyOption; -import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -82,7 +78,6 @@ import java.util.Optional; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Predicate; import static org.apache.fluss.utils.Preconditions.checkState; import static org.apache.fluss.utils.concurrent.LockUtils.inLock; @@ -395,9 +390,7 @@ private void cleanupStaleKvDirectories() { for (File dataDir : dataDirs) { try { Path realDataDir = dataDir.toPath().toRealPath(); - List staleDirs = - listTabletsToLoad( - realDataDir.toFile(), this::listCleanupDirectories); + List staleDirs = listTabletsToLoad(realDataDir.toFile()); int deletedDirectories = 0; for (File tabletDir : staleDirs) { try { @@ -449,34 +442,6 @@ private void deleteStaleKvDirectory(Path tabletDir, Path dataDir) throws IOExcep deleteEmptyParentDirectories(deletedDir.getParent(), dataDir); } - private List listCleanupDirectories(File parent, Predicate nameFilter) - throws IOException { - List directories = new ArrayList<>(); - try (DirectoryStream entries = Files.newDirectoryStream(parent.toPath())) { - for (Path entry : entries) { - if (!nameFilter.test(entry.getFileName().toString())) { - continue; - } - BasicFileAttributes attributes = - Files.readAttributes( - entry, BasicFileAttributes.class, LinkOption.NOFOLLOW_LINKS); - if (attributes.isSymbolicLink()) { - LOG.warn( - "Skipping symbolic link {} during stale KV cleanup; its target will " - + "not be cleaned.", - entry); - continue; - } - if (attributes.isDirectory()) { - directories.add(entry.toFile()); - } - } - } catch (DirectoryIteratorException e) { - throw e.getCause(); - } - return directories; - } - private void deleteEmptyParentDirectories(Path directory, Path dataDir) throws IOException { for (Path parent = directory; parent != null && parent.startsWith(dataDir) && !parent.equals(dataDir); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java b/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java index 8e0997243a..2ed50fa1c7 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java @@ -48,23 +48,10 @@ final class TabletManagerBaseTest { @TempDir private File tempDir; - @Test - void testIgnoresHistoricalLookupCacheDirectoryWhenLoadingTablets() { - File fakeTabletDir = - new File( - new File(FlussPaths.historicalLookupRootDir(tempDir), "database"), - "kv-table-1"); - assertThat(fakeTabletDir.mkdirs()).isTrue(); - - TestingTabletManager tabletManager = new TestingTabletManager(tempDir); - - assertThat(tabletManager.tabletsToLoad(tempDir)).isEmpty(); - } - @ParameterizedTest @EnumSource(TabletManagerBase.TabletType.class) - void testListsTabletLayoutAndExcludesCaches(TabletManagerBase.TabletType tabletType) - throws Exception { + void testListsTabletLayoutAndExcludesCachesAndSymbolicLinks( + TabletManagerBase.TabletType tabletType, @TempDir Path outsideDir) throws Exception { for (String path : Arrays.asList( "db/table-1/kv-0", @@ -78,6 +65,10 @@ void testListsTabletLayoutAndExcludesCaches(TabletManagerBase.TabletType tabletT FlussPaths.REMOTE_LOG_INDEX_LOCAL_CACHE + "/table-1/log-0")) { Files.createDirectories(tempDir.toPath().resolve(path)); } + Files.createDirectories(outsideDir.resolve("table-3/kv-0")); + Files.createDirectories(outsideDir.resolve("table-3/log-0")); + Files.createSymbolicLink(tempDir.toPath().resolve("linked-db"), outsideDir); + String prefix = tabletType == TabletManagerBase.TabletType.KV ? FlussPaths.KV_TABLET_DIR_PREFIX @@ -90,18 +81,6 @@ void testListsTabletLayoutAndExcludesCaches(TabletManagerBase.TabletType tabletT new File(tempDir, "db/table-2/partition-p2/" + prefix + "1")); } - @Test - void testLogLoadingStillFollowsSymbolicDatabaseDirectory(@TempDir Path outsideDir) - throws Exception { - Files.createDirectories(outsideDir.resolve("table-1/log-0")); - Path dbLink = Files.createSymbolicLink(tempDir.toPath().resolve("db"), outsideDir); - TestingTabletManager tabletManager = - new TestingTabletManager(tempDir, TabletManagerBase.TabletType.LOG); - - assertThat(tabletManager.tabletsToLoad(tempDir)) - .containsExactly(dbLink.resolve("table-1/log-0").toFile()); - } - @Test void testCloseTabletsConcurrentlyWaitsForAllTasksAndShutsDownPoolOnFailure() throws Exception { TestingTabletManager tabletManager = new TestingTabletManager(2); @@ -155,10 +134,6 @@ private TestingTabletManager(int closingThreads) { super(TabletType.KV, Collections.emptyList(), new Configuration(), closingThreads); } - private TestingTabletManager(File dataDir) { - this(dataDir, TabletType.KV); - } - private TestingTabletManager(File dataDir, TabletType tabletType) { super(tabletType, Collections.singletonList(dataDir), new Configuration(), 1); } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java index 31f92f5432..2d7c7eaa51 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java @@ -63,9 +63,7 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.MethodSource; -import org.junit.jupiter.params.provider.ValueSource; import javax.annotation.Nullable; @@ -194,7 +192,9 @@ void testStartupCleanupAcrossDisks(@TempDir File secondDataDir) throws Exception "db/table-1/kv-0", "db/table-2/20260917-p2/kv-1", "dropped/table-3/kv-0", - "dropped/table-4/20260917-p4/kv-2")) { + "dropped/table-4/20260917-p4/kv-2", + "dropped/table-3/kv-0.old.deleted", + "dropped/no-id/kv-invalid")) { Path kvDir = dataDir.toPath().resolve(path); Files.createDirectories(kvDir.resolve("db")); Files.write(kvDir.resolve("db/000001.sst"), new byte[] {1, 2, 3}); @@ -249,23 +249,7 @@ private void configureTwoDataDirs(File secondDataDir) throws Exception { } @Test - void testStartupCleanupRejectsOpenTablets() throws Exception { - initTableBuckets(null); - KvTablet kv = getOrCreateKv(tablePath1, null, tableBucket1); - byte[] key = "live-key".getBytes(StandardCharsets.UTF_8); - KvRecord record = kvRecordFactory.ofRecord(key, new Object[] {1, "value"}); - put(kv, record); - - assertThatThrownBy(kvManager::startup) - .isInstanceOf(IllegalStateException.class) - .hasMessage("Cannot clean KV directories while KV tablets are open."); - - assertThat(kv.getKvTabletDir()).isDirectory(); - verifyMultiGet(kv, key, valueOf(record)); - } - - @Test - void testStartupCleanupContinuesOnOtherDisksAfterDeletionFailure(@TempDir File secondDataDir) + void testStartupCleanupIsolatesFailedDeletionsAndRetries(@TempDir File secondDataDir) throws Exception { configureTwoDataDirs(secondDataDir); Path otherKvDir = @@ -286,39 +270,22 @@ void testStartupCleanupContinuesOnOtherDisksAfterDeletionFailure(@TempDir File s assertThat(Files.readAllBytes(pendingDirs[0].toPath().resolve("db/data"))) .containsExactly(1); assertThat(otherKvDir).doesNotExist(); - } finally { - makeKvStoreDirectoriesWritable(tableDir); - } - - // Cleanup can be retried once the disk problem is resolved. - kvManager.startup(); - assertThat(tableDir).doesNotExist(); - } - @Test - void testStartupCleanupRetriesPendingDeletionWithoutBlockingNewTablet() throws Exception { - Path kvDir = Files.createDirectories(tempDir.toPath().resolve("db/table-1/kv-0")); - Files.write(kvDir.resolve("data"), new byte[] {1}); - Path pendingDir = Files.createDirectory(kvDir.resolveSibling("kv-0.deleted")); - Path pendingFile = Files.write(pendingDir.resolve("data"), new byte[] {2}); - try { - assertThat(pendingDir.toFile().setWritable(false)).isTrue(); - assumeThat(Files.isWritable(pendingDir)).isFalse(); - - kvManager.startup(); - // Retrying an older deletion must neither rename it again nor prevent isolation of - // the current tablet, regardless of directory enumeration order. + // A pending deletion must not block cleanup of a new tablet at the original path. + Files.createDirectories(kvDir.resolve("db")); + Files.write(kvDir.resolve("db/data"), new byte[] {2}); kvManager.startup(); - assertThat(kvDir).doesNotExist(); - assertThat(kvDir.getParent().toFile().listFiles()).containsExactly(pendingDir.toFile()); - assertThat(Files.readAllBytes(pendingFile)).containsExactly(2); + assertThat(tableDir.listFiles()).containsExactly(pendingDirs[0]); + assertThat(Files.readAllBytes(pendingDirs[0].toPath().resolve("db/data"))) + .containsExactly(1); } finally { - assertThat(pendingDir.toFile().setWritable(true)).isTrue(); + makeKvStoreDirectoriesWritable(tableDir); } + // Cleanup can be retried once the disk problem is resolved. kvManager.startup(); - assertThat(kvDir.getParent()).doesNotExist(); + assertThat(tableDir).doesNotExist(); } @Test @@ -357,65 +324,22 @@ private void makeKvStoreDirectoriesWritable(File directory) { } @Test - void testStartupCleanupWithSymbolicDataRoot(@TempDir Path linkDir) throws Exception { - tearDown(); - kvManager = null; - logManager = null; - localDiskManager = null; - Path dataLink = Files.createSymbolicLink(linkDir.resolve("data"), tempDir.toPath()); - conf.set(ConfigOptions.DATA_DIR, dataLink.toString()); - createManagers(); - Path staleDir = tempDir.toPath().resolve("db/table-1/kv-0"); - Files.createDirectories(staleDir); - Files.write(staleDir.resolve("data"), new byte[] {1}); - - kvManager.startup(); - - assertThat(staleDir).doesNotExist(); - assertThat(Files.isSymbolicLink(dataLink)).isTrue(); - assertThat(tempDir).isDirectory(); - } - - @ParameterizedTest - @CsvSource({ - "db, table-1/kv-0", - "db/table-1, kv-0", - "db/table-1/partition-p1, kv-0", - "db/table-1/kv-0, db", - "db/table-1/partition-p1/kv-0, db" - }) - void testStartupCleanupSkipsSymbolicLinks( - String linkPath, String targetPath, @TempDir Path outsideDir) throws Exception { - Path outsideKvDir = Files.createDirectories(outsideDir.resolve(targetPath)); - Path outsideFile = Files.write(outsideKvDir.resolve("data"), new byte[] {1, 2, 3}); - Path link = tempDir.toPath().resolve(linkPath); - Files.createDirectories(link.getParent()); - Files.createSymbolicLink(link, outsideDir); - Path staleDir = Files.createDirectories(tempDir.toPath().resolve("other/table-2/kv-0")); - - kvManager.startup(); - - assertThat(staleDir).doesNotExist(); - assertThat(Files.readAllBytes(outsideFile)).containsExactly(1, 2, 3); - assertThat(Files.isSymbolicLink(link)).isTrue(); - } - - @ParameterizedTest - @ValueSource(strings = {"db", "db/table-1", "db/table-1/partition-p1"}) - void testStartupCleanupContinuesOnOtherDisksAfterListingFailure( - String unreadablePath, @TempDir File secondDataDir) throws Exception { + void testStartupCleanupSkipsUnreadableDirectories(@TempDir File secondDataDir) + throws Exception { configureTwoDataDirs(secondDataDir); Path otherKvDir = Files.createDirectories(secondDataDir.toPath().resolve("db/table-2/kv-0")); Path kvDir = tempDir.toPath().resolve("db/table-1/partition-p1/kv-0"); Files.createDirectories(kvDir); Path retainedFile = Files.write(kvDir.resolve("data"), new byte[] {1}); - File unreadableDir = tempDir.toPath().resolve(unreadablePath).toFile(); + Path siblingKvDir = Files.createDirectories(tempDir.toPath().resolve("db/table-3/kv-0")); + File unreadableDir = tempDir.toPath().resolve("db/table-1").toFile(); try { assertThat(unreadableDir.setReadable(false)).isTrue(); assumeThat(unreadableDir.canRead()).isFalse(); kvManager.startup(); + assertThat(siblingKvDir).doesNotExist(); assertThat(otherKvDir).doesNotExist(); } finally { assertThat(unreadableDir.setReadable(true)).isTrue(); @@ -426,55 +350,6 @@ void testStartupCleanupContinuesOnOtherDisksAfterListingFailure( assertThat(kvDir).doesNotExist(); } - @Test - void testStartupCleanupLeavesExcludedSymbolicLinks(@TempDir Path outsideDir) throws Exception { - Path outsideFile = Files.write(outsideDir.resolve("data"), new byte[] {1, 2, 3}); - Path kvDir = tempDir.toPath().resolve("db/table-1/kv-0"); - Files.createDirectories(kvDir); - List links = new ArrayList<>(); - for (String path : - Arrays.asList( - FlussPaths.HISTORICAL_LOOKUP_CACHE_DIR_NAME, - FlussPaths.REMOTE_LOG_INDEX_LOCAL_CACHE, - "db/table-1/log-0", - "db/table-1/backup")) { - links.add(Files.createSymbolicLink(tempDir.toPath().resolve(path), outsideDir)); - } - - kvManager.startup(); - - assertThat(kvDir).doesNotExist(); - assertThat(Files.readAllBytes(outsideFile)).containsExactly(1, 2, 3); - for (Path link : links) { - assertThat(Files.isSymbolicLink(link)).isTrue(); - } - } - - @ParameterizedTest - @ValueSource(strings = {"db/foo/kv-0", "db/table-1/kv-abc", "db/foo/partition-p1/kv-abc"}) - void testStartupCleanupDoesNotRequireValidTabletIds(String path) throws Exception { - Path staleDir = Files.createDirectories(tempDir.toPath().resolve(path)); - Files.write(staleDir.resolve("data"), new byte[] {1}); - - kvManager.startup(); - - assertThat(staleDir).doesNotExist(); - assertThat(tempDir.toPath().resolve("db")).doesNotExist(); - assertThat(tempDir).isDirectory(); - } - - @Test - void testStartupCleanupDoesNotFollowLinksInsideKv(@TempDir Path outsideDir) throws Exception { - Path outsideFile = Files.write(outsideDir.resolve("data"), new byte[] {1, 2, 3}); - Path staleDir = Files.createDirectories(tempDir.toPath().resolve("db/table-1/kv-0")); - Files.createSymbolicLink(staleDir.resolve("db"), outsideDir); - - kvManager.startup(); - - assertThat(staleDir).doesNotExist(); - assertThat(Files.readAllBytes(outsideFile)).containsExactly(1, 2, 3); - } - @Test void testPositiveSharedBlockCacheSizeEnablesSharedCache() throws Exception { kvManager.shutdown(); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java index d1921f7df4..2f2bd1e19c 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java @@ -33,13 +33,11 @@ import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.LeaderAndIsr; import org.apache.fluss.types.DataTypes; -import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.types.Tuple2; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import org.rocksdb.FlushOptions; @@ -68,7 +66,6 @@ import static org.apache.fluss.testutils.DataTestUtils.getKeyValuePairs; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assumptions.assumeThat; /** The ITCase for tabletServer shutdown (controlled shutdown). */ public class TabletServerShutdownITCase { @@ -264,12 +261,9 @@ void testControlledShutdownCleansKvWhenRestartedAsFollower() throws Exception { } @ParameterizedTest - @CsvSource({"true, false", "false, false", "true, true", "false, true"}) - void testKvRecoveryAfterStartupCleanup(boolean withSnapshot, boolean failCleanup) - throws Exception { - TablePath tablePath = - TablePath.of( - "test_shutdown", "kv_startup_recovery_" + withSnapshot + "_" + failCleanup); + @ValueSource(booleans = {true, false}) + void testKvRecoveryAfterStartupCleanup(boolean withSnapshot) throws Exception { + TablePath tablePath = TablePath.of("test_shutdown", "kv_startup_recovery_" + withSnapshot); long tableId = createTable( FLUSS_CLUSTER_EXTENSION, @@ -306,57 +300,19 @@ void testKvRecoveryAfterStartupCleanup(boolean withSnapshot, boolean failCleanup Files.write(staleFile, new byte[] {1}); FLUSS_CLUSTER_EXTENSION.stopTabletServer(leader); - boolean restarted = false; - try { - if (failCleanup) { - // The tablet can be renamed, but recursive deletion of its RocksDB files fails. - File dbDir = new File(kvDir, "db"); - assertThat(dbDir.setWritable(false)).isTrue(); - assumeThat(Files.isWritable(dbDir.toPath())).isFalse(); - } - FLUSS_CLUSTER_EXTENSION.startTabletServer(leader); - restarted = true; - Replica recovered = FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tableBucket); - assertThat(staleFile).doesNotExist(); - assertThat(kvDir).isDirectory(); - if (failCleanup) { - File[] pendingDeletionDirs = - kvDir.getParentFile() - .listFiles( - file -> - file.getName().startsWith(kvDir.getName() + ".") - && file.getName() - .endsWith( - FlussPaths - .DELETED_FILE_SUFFIX)); - assertThat(pendingDeletionDirs).hasSize(1); - assertThat(new File(pendingDeletionDirs[0], "db")).isDirectory(); - } - assertThat(recovered.getRowCount()).isEqualTo(2L); - TabletServerGateway recoveredGateway = - FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leader); - for (Tuple2 keyValue : - getKeyValuePairs( - genKvRecords(new Object[] {1, "updated"}, new Object[] {2, "b1"}))) { - assertLookupResponse( - recoveredGateway.lookup(newLookupRequest(tableId, 0, keyValue.f0)).get(), - keyValue.f1); - } - } finally { - if (failCleanup) { - // Restore permissions at either the original path or the renamed deletion path. - File[] tabletDirs = kvDir.getParentFile().listFiles(File::isDirectory); - assertThat(tabletDirs).isNotNull(); - for (File tabletDir : tabletDirs) { - File dbDir = new File(tabletDir, "db"); - if (dbDir.exists()) { - assertThat(dbDir.setWritable(true)).isTrue(); - } - } - } - if (!restarted) { - FLUSS_CLUSTER_EXTENSION.startTabletServer(leader); - } + FLUSS_CLUSTER_EXTENSION.startTabletServer(leader); + Replica recovered = FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tableBucket); + assertThat(staleFile).doesNotExist(); + assertThat(kvDir).isDirectory(); + assertThat(recovered.getRowCount()).isEqualTo(2L); + TabletServerGateway recoveredGateway = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leader); + for (Tuple2 keyValue : + getKeyValuePairs( + genKvRecords(new Object[] {1, "updated"}, new Object[] {2, "b1"}))) { + assertLookupResponse( + recoveredGateway.lookup(newLookupRequest(tableId, 0, keyValue.f0)).get(), + keyValue.f1); } dropTable(FLUSS_CLUSTER_EXTENSION, tablePath); }