natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Stefan Miklosovic 2023-02-17 15:44:17 +01:00
parent d7352209b2 cfe9641fbe
commit 8d54835d5b
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
10 changed files with 242 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -106,6 +106,7 @@
* Add guardrail for ALTER TABLE ADD / DROP / REMOVE column operations (CASSANDRA-17495)
* Rename DisableFlag class to EnableFlag on guardrails (CASSANDRA-17544)
Merged from 4.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)

6
src/java/org/apache/cassandra/db/Directories.java
View File

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

12
src/java/org/apache/cassandra/io/util/FileUtils.java
View File

@ -488,6 +488,9 @@ public final class FileUtils
*/
public static long folderSize(File folder)
{
if (!folder.exists())
return 0;
final long [] sizeArr = {0L};
try
{
@ -499,6 +502,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)

20
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -74,7 +74,6 @@ import org.apache.cassandra.metrics.Sampler;
import org.apache.cassandra.metrics.SamplingManager;
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;
@ -1061,10 +1060,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));
@ -4340,10 +4345,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
*/
private void clearKeyspaceSnapshot(String keyspace, String tag, long olderThanTimestamp)
{
Set<TableSnapshot> snapshotsToClear = new SnapshotLoader().loadSnapshots(keyspace)
.stream()
.filter(TableSnapshot.shouldClearSnapshot(tag, olderThanTimestamp))
.collect(Collectors.toSet());
Set<TableSnapshot> snapshotsToClear = snapshotManager.loadSnapshots(keyspace)
.stream()
.filter(TableSnapshot.shouldClearSnapshot(tag, olderThanTimestamp))
.collect(Collectors.toSet());
for (TableSnapshot snapshot : snapshotsToClear)
snapshotManager.clearSnapshot(snapshot);
}
@ -4353,10 +4358,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
boolean skipExpiring = options != null && Boolean.parseBoolean(options.getOrDefault("no_ttl", "false"));
boolean includeEphemeral = options != null && Boolean.parseBoolean(options.getOrDefault("include_ephemeral", "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;

152
src/java/org/apache/cassandra/service/snapshot/SnapshotLoader.java
View File

@ -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,15 +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>.+)$");
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 Collection<Path> dataDirectories;
private final Map<String, TableSnapshot.Builder> snapshots = new HashMap<>();
public SnapshotLoader()
{
@ -80,11 +79,90 @@ public class SnapshotLoader extends SimpleFileVisitor<Path>
this(directories.getCFDirectories().stream().map(File::toPath).collect(Collectors.toList()));
}
@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(String keyspace)
{
// if we supply a keyspace, the walking max depth will be suddenly shorther
// because we are one level down in the directory structure
int maxDepth = keyspace == null ? 5 : 4;
Map<String, TableSnapshot.Builder> snapshots = new HashMap<>();
Visitor visitor = new Visitor(snapshots);
for (Path dataDir : dataDirectories)
{
if (keyspace != null)
@ -93,19 +171,16 @@ public class SnapshotLoader extends SimpleFileVisitor<Path>
try
{
if (new File(dataDir).exists())
{
Files.walkFileTree(dataDir, Collections.emptySet(), maxDepth, this);
}
Files.walkFileTree(dataDir, Collections.emptySet(), maxDepth, visitor);
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());
}
@ -113,65 +188,4 @@ public class SnapshotLoader extends SimpleFileVisitor<Path>
{
return loadSnapshots(null);
}
@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 = UUID_PATTERN.matcher(uuidWithoutDashes).replaceFirst("$1-$2-$3-$4-$5");
return UUID.fromString(dashedUUID);
}
}

53
src/java/org/apache/cassandra/service/snapshot/SnapshotManager.java
View File

@ -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,26 @@ public class SnapshotManager {
}
}
@VisibleForTesting
protected synchronized void loadSnapshots()
public synchronized Set<TableSnapshot> loadSnapshots(String keyspace)
{
SnapshotLoader loader = new SnapshotLoader(DatabaseDescriptor.getAllDataFileLocations());
addSnapshots(loader.loadSnapshots());
return snapshotLoader.loadSnapshots(keyspace);
}
public synchronized Set<TableSnapshot> loadSnapshots()
{
return snapshotLoader.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 +142,25 @@ public class SnapshotManager {
@VisibleForTesting
protected synchronized void clearExpiredSnapshots()
{
Instant now = now();
while (!expiringSnapshots.isEmpty() && expiringSnapshots.peek().isExpired(now))
TableSnapshot expiredSnapshot;
while ((expiredSnapshot = expiringSnapshots.peek()) != null)
{
TableSnapshot expiredSnapshot = expiringSnapshots.peek();
if (expiredSnapshot != null)
{
logger.debug("Removing expired snapshot {}.", expiredSnapshot);
clearSnapshot(expiredSnapshot);
}
if (!expiredSnapshot.isExpired(now()))
break; // the earliest expiring snapshot is not expired yet, so there is no more expired snapshots to remove
logger.debug("Removing expired snapshot {}.", expiredSnapshot);
clearSnapshot(expiredSnapshot);
}
}
/**
* Deletes snapshot and remove it from manager
*/
public void clearSnapshot(TableSnapshot snapshot)
public synchronized void clearSnapshot(TableSnapshot snapshot)
{
for (File snapshotDir : snapshot.getDirectories())
{
Directories.removeSnapshotDirectory(DatabaseDescriptor.getSnapshotRateLimiter(), snapshotDir);
}
expiringSnapshots.remove(snapshot);
}

1
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -709,6 +709,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,

55
test/distributed/org/apache/cassandra/distributed/test/SnapshotsTest.java
View File

@ -18,23 +18,23 @@
package org.apache.cassandra.distributed.test;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
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 java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
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(() -> {
NodeToolResult listsnapshots;
if (noTTL)
listsnapshots = cluster.get(1).nodetoolResult("listsnapshots", "-nt");
else
listsnapshots = cluster.get(1).nodetoolResult("listsnapshots");
.until(() -> waitForSnapshotInternal(snapshotName, expectPresent, noTTL));
}
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());
private boolean waitForSnapshotInternal(String snapshotName, boolean expectPresent, boolean noTTL) {
NodeToolResult listsnapshots;
if (noTTL)
listsnapshots = cluster.get(1).nodetoolResult("listsnapshots", "-nt");
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));
}
}

2
test/unit/org/apache/cassandra/service/snapshot/SnapshotLoaderTest.java
View File

@ -269,7 +269,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

82
test/unit/org/apache/cassandra/service/snapshot/SnapshotManagerTest.java
View File

@ -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,17 +57,23 @@ public class SnapshotManagerTest
@ClassRule
public static TemporaryFolder temporaryFolder = new TemporaryFolder();
private TableSnapshot generateSnapshotDetails(String tag, Instant expiration, boolean ephemeral) throws Exception {
return new TableSnapshot(
"ks",
"tbl",
UUID.randomUUID(),
tag,
Instant.EPOCH,
expiration,
createFolders(temporaryFolder),
ephemeral
);
private TableSnapshot generateSnapshotDetails(String tag, Instant expiration, boolean ephemeral)
{
try
{
return new TableSnapshot("ks",
"tbl",
UUID.randomUUID(),
tag,
Instant.EPOCH,
expiration,
createFolders(temporaryFolder),
ephemeral);
}
catch (Exception ex)
{
throw new RuntimeException(ex);
}
}
@Test
@ -163,4 +176,49 @@ public class SnapshotManagerTest
assertThat(manager.getExpiringSnapshots()).doesNotContain(expiringSnapshot);
assertThat(expiringSnapshot.exists()).isFalse();
}
}
@Test // see CASSANDRA-18211
public void testConcurrentClearingOfSnapshots() throws Exception
{
AtomicReference<Long> firstInvocationTime = new AtomicReference<>(0L);
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), false);
manager.addSnapshot(expiringSnapshot);
manager.resumeSnapshotCleanup();
Thread nonExpiringSnapshotCleanupThred = new Thread(() -> manager.clearSnapshot(generateSnapshotDetails("mysnapshot2", null, false)));
// 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);
}
}