Fix possible NoSuchFileException when removing a snapshot

patch by Stefan Miklosovic; reviewed by Jacek Lewandowski for CASSANDRA-18211

Co-authored-by: Jacek Lewandowski <lewandowski.jacek@gmail.com>

CHANGES.txt

@@ -1,4 +1,5 @@
 4.1.1
+ * Fix possible NoSuchFileException when removing a snapshot (CASSANDRA-18211)
  * PaxosPrepare may add instances to the Electorate that are not in gossip (CASSANDRA-18194)
  * Fix PAXOS2_COMMIT_AND_PREPARE_RSP serialisation AssertionError (CASSANDRA-18164)
  * Streaming progress virtual table lock contention can trigger TCP_USER_TIMEOUT and fail streaming (CASSANDRA-18110)

src/java/org/apache/cassandra/db/Directories.java

@@ -1119,9 +1119,11 @@ public class Directories
             {
                 FileUtils.deleteRecursiveWithThrottle(snapshotDir, snapshotRateLimiter);
             }
-            catch (FSWriteError e)
+            catch (RuntimeException ex)
             {
-                throw e;
+                if (!snapshotDir.exists())
+                    return; // ignore
+                throw ex;
             }
         }
     }

src/java/org/apache/cassandra/io/util/FileUtils.java

@@ -489,6 +489,9 @@ public final class FileUtils
      */
     public static long folderSize(File folder)
     {
+        if (!folder.exists())
+            return 0;
+
         final long [] sizeArr = {0L};
         try
         {
@@ -500,6 +503,15 @@ public final class FileUtils
                     sizeArr[0] += attrs.size();
                     return FileVisitResult.CONTINUE;
                 }
+
+                @Override
+                public FileVisitResult visitFileFailed(Path path, IOException e) throws IOException
+                {
+                    if (e instanceof NoSuchFileException)
+                        return FileVisitResult.CONTINUE;
+                    else
+                        throw e;
+                }
             });
         }
         catch (IOException e)

src/java/org/apache/cassandra/service/StorageService.java

@@ -72,7 +72,6 @@ import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
 import org.apache.cassandra.schema.Keyspaces;
 import org.apache.cassandra.service.disk.usage.DiskUsageBroadcaster;
-import org.apache.cassandra.service.snapshot.SnapshotLoader;
 import org.apache.cassandra.utils.concurrent.Future;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.utils.concurrent.FutureCombiner;
@@ -1041,10 +1040,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
             DiskUsageBroadcaster.instance.startBroadcasting();
             HintsService.instance.startDispatch();
             BatchlogManager.instance.start();
-            snapshotManager.start();
+            startSnapshotManager();
         }
     }
 
+    @VisibleForTesting
+    public void startSnapshotManager()
+    {
+        snapshotManager.start();
+    }
+
     public void waitForSchema(long schemaTimeoutMillis, long ringTimeoutMillis)
     {
         Instant deadline = FBUtilities.now().plus(java.time.Duration.ofMillis(ringTimeoutMillis));
@@ -4222,10 +4227,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     {
         boolean skipExpiring = options != null && Boolean.parseBoolean(options.getOrDefault("no_ttl", "false"));
 
-        SnapshotLoader loader = new SnapshotLoader();
         Map<String, TabularData> snapshotMap = new HashMap<>();
 
-        for (TableSnapshot snapshot : loader.loadSnapshots())
+        for (TableSnapshot snapshot : snapshotManager.loadSnapshots())
         {
             if (skipExpiring && snapshot.isExpiring())
                 continue;

src/java/org/apache/cassandra/service/snapshot/SnapshotLoader.java

@@ -37,6 +37,7 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -50,14 +51,13 @@ import static org.apache.cassandra.service.snapshot.TableSnapshot.buildSnapshotI
 /**
  * Loads snapshot metadata from data directories
  */
-public class SnapshotLoader extends SimpleFileVisitor<Path>
+public class SnapshotLoader
 {
     private static final Logger logger = LoggerFactory.getLogger(SnapshotLoader.class);
 
-    static final Pattern SNAPSHOT_DIR_PATTERN = Pattern.compile("(?<keyspace>\\w+)/(?<tableName>\\w+)\\-(?<tableId>[0-9a-f]{32})/snapshots/(?<tag>[\\w-]+)$");
+    static final Pattern SNAPSHOT_DIR_PATTERN = Pattern.compile("(?<keyspace>\\w+)/(?<tableName>\\w+)-(?<tableId>[0-9a-f]{32})/snapshots/(?<tag>[\\w-]+)$");
 
     private final Collection<Path> dataDirectories;
-    private final Map<String, TableSnapshot.Builder> snapshots = new HashMap<>();
 
     public SnapshotLoader()
     {
@@ -74,87 +74,101 @@ public class SnapshotLoader extends SimpleFileVisitor<Path>
         this.dataDirectories = dataDirs;
     }
 
+    @VisibleForTesting
+    static class Visitor extends SimpleFileVisitor<Path>
+    {
+        private static final Pattern UUID_PATTERN = Pattern.compile("([0-9a-f]{8})([0-9a-f]{4})([0-9a-f]{4})([0-9a-f]{4})([0-9a-f]+)");
+        private final Map<String, TableSnapshot.Builder> snapshots;
+
+        public Visitor(Map<String, TableSnapshot.Builder> snapshots)
+        {
+            this.snapshots = snapshots;
+        }
+
+        @Override
+        public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException
+        {
+            // Cassandra can remove some files while traversing the tree,
+            // for example when SSTables are compacted while we are walking it.
+            // SnapshotLoader is interested only in SSTables in snapshot directories which are not compacted,
+            // but we need to cover these in regular table directories too.
+            // If listing failed but exception is NoSuchFileException, then we
+            // just skip it and continue with the listing.
+            if (exc instanceof NoSuchFileException)
+                return FileVisitResult.CONTINUE;
+            else
+                throw exc;
+        }
+
+        @Override
+        public FileVisitResult preVisitDirectory(Path subdir, BasicFileAttributes attrs)
+        {
+            if (subdir.getParent().getFileName().toString().equals(SNAPSHOT_SUBDIR))
+            {
+                logger.trace("Processing directory " + subdir);
+                Matcher snapshotDirMatcher = SNAPSHOT_DIR_PATTERN.matcher(subdir.toString());
+                if (snapshotDirMatcher.find())
+                {
+                    try
+                    {
+                        loadSnapshotFromDir(snapshotDirMatcher, subdir);
+                    }
+                    catch (Throwable e)
+                    {
+                        logger.warn("Could not load snapshot from {}.", subdir, e);
+                    }
+                }
+                return FileVisitResult.SKIP_SUBTREE;
+            }
+
+            return subdir.getFileName().toString().equals(Directories.BACKUPS_SUBDIR)
+                   ? FileVisitResult.SKIP_SUBTREE
+                   : FileVisitResult.CONTINUE;
+        }
+
+        /**
+         * Given an UUID string without dashes (ie. c7e513243f0711ec9bbc0242ac130002)
+         * return an UUID object (ie. c7e51324-3f07-11ec-9bbc-0242ac130002)
+         */
+        static UUID parseUUID(String uuidWithoutDashes) throws IllegalArgumentException
+        {
+            assert uuidWithoutDashes.length() == 32 && !uuidWithoutDashes.contains("-");
+            String dashedUUID = UUID_PATTERN.matcher(uuidWithoutDashes).replaceFirst("$1-$2-$3-$4-$5");
+            return UUID.fromString(dashedUUID);
+        }
+
+        private void loadSnapshotFromDir(Matcher snapshotDirMatcher, Path snapshotDir)
+        {
+            String keyspaceName = snapshotDirMatcher.group("keyspace");
+            String tableName = snapshotDirMatcher.group("tableName");
+            UUID tableId = parseUUID(snapshotDirMatcher.group("tableId"));
+            String tag = snapshotDirMatcher.group("tag");
+            String snapshotId = buildSnapshotId(keyspaceName, tableName, tableId, tag);
+            TableSnapshot.Builder builder = snapshots.computeIfAbsent(snapshotId, k -> new TableSnapshot.Builder(keyspaceName, tableName, tableId, tag));
+            builder.addSnapshotDir(new File(snapshotDir));
+        }
+    }
+
     public Set<TableSnapshot> loadSnapshots()
     {
+        Map<String, TableSnapshot.Builder> snapshots = new HashMap<>();
+        Visitor visitor = new Visitor(snapshots);
+
         for (Path dataDir : dataDirectories)
         {
             try
             {
                 if (new File(dataDir).exists())
-                {
+                    Files.walkFileTree(dataDir, Collections.emptySet(), 5, visitor);
-                    Files.walkFileTree(dataDir, Collections.EMPTY_SET, 5, this);
-                }
                 else
-                {
                     logger.debug("Skipping non-existing data directory {}", dataDir);
-                }
             }
             catch (IOException e)
             {
                 throw new RuntimeException(String.format("Error while loading snapshots from %s", dataDir), e);
             }
         }
+
         return snapshots.values().stream().map(TableSnapshot.Builder::build).collect(Collectors.toSet());
     }
-
-    @Override
-    public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
-        // Cassandra can remove some files while traversing the tree,
-        // for example when SSTables are compacted while we are walking it.
-        // SnapshotLoader is interested only in SSTables in snapshot directories which are not compacted,
-        // but we need to cover these in regular table directories too.
-        // If listing failed but such file exists and the exception is not NoSuchFileException, then we
-        // have a legitimate error while traversing the tree, otherwise just skip it and continue with the listing.
-        if (Files.exists(file) && !(exc instanceof NoSuchFileException))
-            return super.visitFileFailed(file, exc);
-        else
-            return FileVisitResult.CONTINUE;
-    }
-
-    @Override
-    public FileVisitResult preVisitDirectory(Path subdir, BasicFileAttributes attrs)
-    {
-        if (subdir.getParent().getFileName().toString().equals(SNAPSHOT_SUBDIR))
-        {
-            logger.trace("Processing directory " + subdir);
-            Matcher snapshotDirMatcher = SNAPSHOT_DIR_PATTERN.matcher(subdir.toString());
-            if (snapshotDirMatcher.find())
-            {
-                try
-                {
-                    loadSnapshotFromDir(snapshotDirMatcher, subdir);
-                } catch (Throwable e)
-                {
-                    logger.warn("Could not load snapshot from {}.", subdir, e);
-                }
-            }
-            return FileVisitResult.SKIP_SUBTREE;
-        }
-
-        return subdir.getFileName().toString().equals(Directories.BACKUPS_SUBDIR)
-               ? FileVisitResult.SKIP_SUBTREE
-               : FileVisitResult.CONTINUE;
-    }
-
-    private void loadSnapshotFromDir(Matcher snapshotDirMatcher, Path snapshotDir)
-    {
-        String keyspaceName = snapshotDirMatcher.group("keyspace");
-        String tableName = snapshotDirMatcher.group("tableName");
-        UUID tableId = parseUUID(snapshotDirMatcher.group("tableId"));
-        String tag = snapshotDirMatcher.group("tag");
-        String snapshotId = buildSnapshotId(keyspaceName, tableName, tableId, tag);
-        TableSnapshot.Builder builder = snapshots.computeIfAbsent(snapshotId, k -> new TableSnapshot.Builder(keyspaceName, tableName, tableId, tag));
-        builder.addSnapshotDir(new File(snapshotDir));
-    }
-
-    /**
-     * Given an UUID string without dashes (ie. c7e513243f0711ec9bbc0242ac130002)
-     * return an UUID object (ie. c7e51324-3f07-11ec-9bbc-0242ac130002)
-     */
-    protected static UUID parseUUID(String uuidWithoutDashes) throws IllegalArgumentException
-    {
-        assert uuidWithoutDashes.length() == 32 && !uuidWithoutDashes.contains("-");
-        String dashedUUID = uuidWithoutDashes.replaceFirst("([0-9a-f]{8})([0-9a-f]{4})([0-9a-f]{4})([0-9a-f]{4})([0-9a-f]+)", "$1-$2-$3-$4-$5");
-        return UUID.fromString(dashedUUID);
-    }
 }

src/java/org/apache/cassandra/service/snapshot/SnapshotManager.java

@@ -18,10 +18,9 @@
 package org.apache.cassandra.service.snapshot;
 
 
-import java.time.Instant;
 import java.util.Collection;
-import java.util.Comparator;
 import java.util.PriorityQueue;
+import java.util.Set;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -34,13 +33,15 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.Directories;
 
 import java.util.concurrent.TimeoutException;
-import java.util.stream.Collectors;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.utils.ExecutorUtils;
+
+import static java.util.Comparator.comparing;
+import static java.util.stream.Collectors.toList;
 import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
 import static org.apache.cassandra.utils.FBUtilities.now;
 
@@ -52,15 +53,16 @@ public class SnapshotManager {
 
     private final long initialDelaySeconds;
     private final long cleanupPeriodSeconds;
+    private final SnapshotLoader snapshotLoader;
 
     @VisibleForTesting
-    protected volatile ScheduledFuture cleanupTaskFuture;
+    protected volatile ScheduledFuture<?> cleanupTaskFuture;
 
     /**
-     * Expiring ssnapshots ordered by expiration date, to allow only iterating over snapshots
+     * Expiring snapshots ordered by expiration date, to allow only iterating over snapshots
      * that need to be removed on {@link this#clearExpiredSnapshots()}
      */
-    private final PriorityQueue<TableSnapshot> expiringSnapshots = new PriorityQueue<>(Comparator.comparing(x -> x.getExpiresAt()));
+    private final PriorityQueue<TableSnapshot> expiringSnapshots = new PriorityQueue<>(comparing(TableSnapshot::getExpiresAt));
 
     public SnapshotManager()
     {
@@ -73,6 +75,7 @@ public class SnapshotManager {
     {
         this.initialDelaySeconds = initialDelaySeconds;
         this.cleanupPeriodSeconds = cleanupPeriodSeconds;
+        snapshotLoader = new SnapshotLoader(DatabaseDescriptor.getAllDataFileLocations());
     }
 
     public Collection<TableSnapshot> getExpiringSnapshots()
@@ -82,7 +85,7 @@ public class SnapshotManager {
 
     public synchronized void start()
     {
-        loadSnapshots();
+        addSnapshots(loadSnapshots());
         resumeSnapshotCleanup();
     }
 
@@ -106,22 +109,21 @@ public class SnapshotManager {
         }
     }
 
-    @VisibleForTesting
+    public synchronized Set<TableSnapshot> loadSnapshots()
-    protected synchronized void loadSnapshots()
     {
-        SnapshotLoader loader = new SnapshotLoader(DatabaseDescriptor.getAllDataFileLocations());
+        return snapshotLoader.loadSnapshots();
-        addSnapshots(loader.loadSnapshots());
     }
 
     @VisibleForTesting
     protected synchronized void addSnapshots(Collection<TableSnapshot> snapshots)
     {
-        logger.debug("Adding snapshots: {}.", Joiner.on(", ").join(snapshots.stream().map(s -> s.getId()).collect(Collectors.toList())));
+        logger.debug("Adding snapshots: {}.", Joiner.on(", ").join(snapshots.stream().map(TableSnapshot::getId).collect(toList())));
         snapshots.forEach(this::addSnapshot);
     }
 
     // TODO: Support pausing snapshot cleanup
-    private synchronized void resumeSnapshotCleanup()
+    @VisibleForTesting
+    synchronized void resumeSnapshotCleanup()
     {
         if (cleanupTaskFuture == null)
         {
@@ -135,27 +137,22 @@ public class SnapshotManager {
     @VisibleForTesting
     protected synchronized void clearExpiredSnapshots()
     {
-        Instant now = now();
+        TableSnapshot expiredSnapshot;
-        while (!expiringSnapshots.isEmpty() && expiringSnapshots.peek().isExpired(now))
+        while ((expiredSnapshot = expiringSnapshots.peek()) != null)
         {
-            TableSnapshot expiredSnapshot = expiringSnapshots.peek();
+            if (!expiredSnapshot.isExpired(now()))
-            if (expiredSnapshot != null)
+                break; // the earliest expiring snapshot is not expired yet, so there is no more expired snapshots to remove
-            {
+
-                logger.debug("Removing expired snapshot {}.", expiredSnapshot);
+            logger.debug("Removing expired snapshot {}.", expiredSnapshot);
-                clearSnapshot(expiredSnapshot);
+            clearSnapshot(expiredSnapshot);
-            }
         }
     }
 
-    /**
+    public synchronized void clearSnapshot(TableSnapshot snapshot)
-     * Deletes snapshot and remove it from manager
-     */
-    protected void clearSnapshot(TableSnapshot snapshot)
     {
         for (File snapshotDir : snapshot.getDirectories())
-        {
             Directories.removeSnapshotDirectory(DatabaseDescriptor.getSnapshotRateLimiter(), snapshotDir);
-        }
+
         expiringSnapshots.remove(snapshot);
     }
 

test/distributed/org/apache/cassandra/distributed/impl/Instance.java

@@ -701,6 +701,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
                     StorageService.instance.setUpDistributedSystemKeyspaces();
                     StorageService.instance.setNormalModeUnsafe();
                     Gossiper.instance.register(StorageService.instance);
+                    StorageService.instance.startSnapshotManager();
                 }
 
                 // Populate tokenMetadata for the second time,

test/distributed/org/apache/cassandra/distributed/test/SnapshotsTest.java

@@ -18,23 +18,23 @@
 
 package org.apache.cassandra.distributed.test;
 
-import org.apache.cassandra.config.CassandraRelevantProperties;
+import java.io.IOException;
-import org.apache.cassandra.distributed.Cluster;
+import java.util.Arrays;
-import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import java.util.List;
-import org.apache.cassandra.distributed.api.Feature;
+import java.util.regex.Pattern;
-import org.apache.cassandra.distributed.api.IInvokableInstance;
+
-import org.apache.cassandra.distributed.api.NodeToolResult;
-import org.apache.cassandra.distributed.shared.WithProperties;
-import org.apache.cassandra.utils.Clock;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
-import java.io.IOException;
+import org.apache.cassandra.config.CassandraRelevantProperties;
-import java.util.Arrays;
+import org.apache.cassandra.distributed.Cluster;
-import java.util.List;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import java.util.regex.Pattern;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.shared.WithProperties;
+import org.apache.cassandra.utils.Clock;
 
 import static java.lang.String.format;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -60,7 +60,7 @@ public class SnapshotsTest extends TestBaseImpl
         properties.set(CassandraRelevantProperties.SNAPSHOT_CLEANUP_INITIAL_DELAY_SECONDS, 0);
         properties.set(CassandraRelevantProperties.SNAPSHOT_CLEANUP_PERIOD_SECONDS, SNAPSHOT_CLEANUP_PERIOD_SECONDS);
         properties.set(CassandraRelevantProperties.SNAPSHOT_MIN_ALLOWED_TTL_SECONDS, FIVE_SECONDS);
-        cluster = init(Cluster.build(1).withConfig(c -> c.with(Feature.GOSSIP)).start());
+        cluster = init(Cluster.build(1).start());
     }
 
     @After
@@ -341,21 +341,24 @@ public class SnapshotsTest extends TestBaseImpl
     private void waitForSnapshot(String snapshotName, boolean expectPresent, boolean noTTL)
     {
         await().timeout(20, SECONDS)
+               .pollDelay(0, SECONDS)
                .pollInterval(1, SECONDS)
-               .until(() -> {
+               .until(() -> waitForSnapshotInternal(snapshotName, expectPresent, noTTL));
-                    NodeToolResult listsnapshots;
+    }
-                    if (noTTL)
-                        listsnapshots = cluster.get(1).nodetoolResult("listsnapshots", "-nt");
-                    else
-                        listsnapshots = cluster.get(1).nodetoolResult("listsnapshots");
 
-                   List<String> lines = Arrays.stream(listsnapshots.getStdout().split("\n"))
+    private boolean waitForSnapshotInternal(String snapshotName, boolean expectPresent, boolean noTTL) {
-                                              .filter(line -> !line.isEmpty())
+        NodeToolResult listsnapshots;
-                                              .filter(line -> !line.startsWith("Snapshot Details:") && !line.startsWith("There are no snapshots"))
+        if (noTTL)
-                                              .filter(line -> !line.startsWith("Snapshot name") && !line.startsWith("Total TrueDiskSpaceUsed"))
+            listsnapshots = cluster.get(1).nodetoolResult("listsnapshots", "-nt");
-                                              .collect(toList());
+        else
+            listsnapshots = cluster.get(1).nodetoolResult("listsnapshots");
 
-                   return expectPresent == lines.stream().anyMatch(line -> line.startsWith(snapshotName));
+        List<String> lines = Arrays.stream(listsnapshots.getStdout().split("\n"))
-                });
+                                   .filter(line -> !line.isEmpty())
+                                   .filter(line -> !line.startsWith("Snapshot Details:") && !line.startsWith("There are no snapshots"))
+                                   .filter(line -> !line.startsWith("Snapshot name") && !line.startsWith("Total TrueDiskSpaceUsed"))
+                                   .collect(toList());
+
+        return expectPresent == lines.stream().anyMatch(line -> line.startsWith(snapshotName));
     }
 }

test/unit/org/apache/cassandra/service/snapshot/SnapshotLoaderTest.java

@@ -203,7 +203,7 @@ public class SnapshotLoaderTest
     @Test
     public void testParseUUID()
     {
-        assertThat(SnapshotLoader.parseUUID("c7e513243f0711ec9bbc0242ac130002")).isEqualTo(UUID.fromString("c7e51324-3f07-11ec-9bbc-0242ac130002"));
+        assertThat(SnapshotLoader.Visitor.parseUUID("c7e513243f0711ec9bbc0242ac130002")).isEqualTo(UUID.fromString("c7e51324-3f07-11ec-9bbc-0242ac130002"));
     }
 
     private void writeManifest(File snapshotDir, Instant creationTime, DurationSpec.IntSecondsBound ttl) throws IOException

test/unit/org/apache/cassandra/service/snapshot/SnapshotManagerTest.java

@@ -22,7 +22,9 @@ import java.time.Instant;
 import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
 
+import com.google.common.util.concurrent.Uninterruptibles;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -32,9 +34,14 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.service.DefaultFSErrorHandler;
 
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.cassandra.service.snapshot.TableSnapshotTest.createFolders;
+import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
 import static org.apache.cassandra.utils.FBUtilities.now;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertTrue;
 
 public class SnapshotManagerTest
 {
@@ -50,20 +57,28 @@ public class SnapshotManagerTest
     @ClassRule
     public static TemporaryFolder temporaryFolder = new TemporaryFolder();
 
-    private TableSnapshot generateSnapshotDetails(String tag, Instant expiration) throws Exception {
+    private TableSnapshot generateSnapshotDetails(String tag, Instant expiration)
-        return new TableSnapshot(
+    {
-        "ks",
+        try
-        "tbl",
+        {
-        UUID.randomUUID(),
+            return new TableSnapshot("ks",
-        tag,
+                                     "tbl",
-        Instant.EPOCH,
+                                     UUID.randomUUID(),
-        expiration,
+                                     tag,
-        createFolders(temporaryFolder)
+                                     Instant.EPOCH,
-        );
+                                     expiration,
+                                     createFolders(temporaryFolder));
+        }
+        catch (Exception ex)
+        {
+            throw new RuntimeException(ex);
+        }
     }
 
+
     @Test
-    public void testLoadSnapshots() throws Exception {
+    public void testLoadSnapshots() throws Exception
+    {
         TableSnapshot expired = generateSnapshotDetails("expired", Instant.EPOCH);
         TableSnapshot nonExpired = generateSnapshotDetails("non-expired", now().plusSeconds(ONE_DAY_SECS));
         TableSnapshot nonExpiring = generateSnapshotDetails("non-expiring", null);
@@ -80,7 +95,8 @@ public class SnapshotManagerTest
     }
 
     @Test
-    public void testClearExpiredSnapshots() throws Exception {
+    public void testClearExpiredSnapshots()
+    {
         SnapshotManager manager = new SnapshotManager(3, 3);
 
         // Add 3 snapshots: expired, non-expired and non-expiring
@@ -109,7 +125,8 @@ public class SnapshotManagerTest
     }
 
     @Test
-    public void testScheduledCleanup() throws Exception {
+    public void testScheduledCleanup() throws Exception
+    {
         SnapshotManager manager = new SnapshotManager(0, 1);
         try
         {
@@ -117,8 +134,7 @@ public class SnapshotManagerTest
             manager.start();
 
             // Add 2 expiring snapshots: one to expire in 2 seconds, another in 1 day
-            int TTL_SECS = 2;
+            TableSnapshot toExpire = generateSnapshotDetails("to-expire", now().plusSeconds(2));
-            TableSnapshot toExpire = generateSnapshotDetails("to-expire", now().plusSeconds(TTL_SECS));
             TableSnapshot nonExpired = generateSnapshotDetails("non-expired", now().plusMillis(ONE_DAY_SECS));
             manager.addSnapshot(toExpire);
             manager.addSnapshot(nonExpired);
@@ -130,11 +146,10 @@ public class SnapshotManagerTest
             assertThat(manager.getExpiringSnapshots()).contains(toExpire);
             assertThat(manager.getExpiringSnapshots()).contains(nonExpired);
 
-            // Sleep 4 seconds
+            await().pollInterval(2, SECONDS)
-            Thread.sleep((TTL_SECS + 2) * 1000L);
+                   .timeout(10, SECONDS)
+                   .until(() -> manager.getExpiringSnapshots().size() == 1);
 
-            // Snapshot with ttl=2s should be gone, while other should remain
-            assertThat(manager.getExpiringSnapshots()).hasSize(1);
             assertThat(manager.getExpiringSnapshots()).contains(nonExpired);
             assertThat(toExpire.exists()).isFalse();
             assertThat(nonExpired.exists()).isTrue();
@@ -145,21 +160,48 @@ public class SnapshotManagerTest
         }
     }
 
-    @Test
+    @Test // see CASSANDRA-18211
-    public void testClearSnapshot() throws Exception
+    public void testConcurrentClearingOfSnapshots() throws Exception
     {
-        // Given
+
-        SnapshotManager manager = new SnapshotManager(1, 3);
+        AtomicReference<Long> firstInvocationTime = new AtomicReference<>(0L);
-        TableSnapshot expiringSnapshot = generateSnapshotDetails("snapshot", now().plusMillis(50000));
+        AtomicReference<Long> secondInvocationTime = new AtomicReference<>(0L);
+
+        SnapshotManager manager = new SnapshotManager(0, 5)
+        {
+            @Override
+            public synchronized void clearSnapshot(TableSnapshot snapshot)
+            {
+                if (snapshot.getTag().equals("mysnapshot"))
+                {
+                    firstInvocationTime.set(currentTimeMillis());
+                    Uninterruptibles.sleepUninterruptibly(10, SECONDS);
+                }
+                else if (snapshot.getTag().equals("mysnapshot2"))
+                {
+                    secondInvocationTime.set(currentTimeMillis());
+                }
+                super.clearSnapshot(snapshot);
+            }
+        };
+
+        TableSnapshot expiringSnapshot = generateSnapshotDetails("mysnapshot", Instant.now().plusSeconds(15));
         manager.addSnapshot(expiringSnapshot);
-        assertThat(manager.getExpiringSnapshots()).contains(expiringSnapshot);
-        assertThat(expiringSnapshot.exists()).isTrue();
 
-        // When
+        manager.resumeSnapshotCleanup();
-        manager.clearSnapshot(expiringSnapshot);
 
-        // Then
+        Thread nonExpiringSnapshotCleanupThred = new Thread(() -> manager.clearSnapshot(generateSnapshotDetails("mysnapshot2", null)));
-        assertThat(manager.getExpiringSnapshots()).doesNotContain(expiringSnapshot);
+
-        assertThat(expiringSnapshot.exists()).isFalse();
+        // wait until the first snapshot expires
+        await().pollInterval(1, SECONDS)
+               .pollDelay(0, SECONDS)
+               .timeout(1, MINUTES)
+               .until(() -> firstInvocationTime.get() > 0);
+
+        // this will block until the first snapshot is cleaned up
+        nonExpiringSnapshotCleanupThred.start();
+        nonExpiringSnapshotCleanupThred.join();
+
+        assertTrue(secondInvocationTime.get() - firstInvocationTime.get() > 10_000);
     }
-}
+}