diff --git a/.gitmodules b/.gitmodules index 616dacf610a7..57ff6dc67b7f 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "modules/accord"] path = modules/accord - url = https://github.com/apache/cassandra-accord.git - branch = trunk + url = https://github.com/alanwang67/cassandra-accord.git + branch = CASSANDRA-19433 diff --git a/modules/accord b/modules/accord index 93d78be37ef9..28d5bca978c7 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 93d78be37ef904118da2426716b5d77c0fb227db +Subproject commit 28d5bca978c75b9c8f4e5ed745e7376395b16df5 diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index ea370b4601fb..11bb019709e8 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -39,6 +39,7 @@ import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Collectors; +import java.util.stream.Stream; import javax.management.openmbean.OpenDataException; import javax.management.openmbean.TabularData; @@ -114,9 +115,11 @@ import org.apache.cassandra.repair.NoSuchRepairSessionException; import org.apache.cassandra.schema.CompactionParams.TombstoneOption; import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ownership.DataPlacement; @@ -133,6 +136,7 @@ import org.apache.cassandra.utils.concurrent.Promise; import org.apache.cassandra.utils.concurrent.Refs; +import static com.google.common.base.Preconditions.checkState; import static java.util.Collections.singleton; import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; import static org.apache.cassandra.concurrent.FutureTask.callable; @@ -580,6 +584,11 @@ private AllSSTableOpStatus parallelAllSSTableOperation(final ColumnFamilyStore c Iterable sstables = Lists.newArrayList(operation.filterSSTables(compacting)); if (Iterables.isEmpty(sstables)) { + if (operation.incompleteOperation()) + { + logger.info("Operation incomplete for {}.{}", keyspace, table); + return AllSSTableOpStatus.INCOMPLETE; + } logger.info("No sstables to {} for {}.{}", operationName, keyspace, table); return AllSSTableOpStatus.SUCCESSFUL; } @@ -612,7 +621,14 @@ public Object call() throws Exception FBUtilities.waitOnFutures(futures); assert compacting.originals().isEmpty(); logger.info("Finished {} for {}.{} successfully", operationType, keyspace, table); - return AllSSTableOpStatus.SUCCESSFUL; + + if (operation.incompleteOperation()) + { + logger.info("Operation incomplete for {}.{}", keyspace, table); + return AllSSTableOpStatus.INCOMPLETE; + } + else + return AllSSTableOpStatus.SUCCESSFUL; } finally { @@ -636,13 +652,18 @@ private static interface OneSSTableOperation { Iterable filterSSTables(LifecycleTransaction transaction); void execute(LifecycleTransaction input) throws IOException; + default boolean incompleteOperation() + { + return false; + } } public enum AllSSTableOpStatus { SUCCESSFUL(0), ABORTED(1), - UNABLE_TO_CANCEL(2); + UNABLE_TO_CANCEL(2), + INCOMPLETE(3); public final int statusCode; @@ -789,24 +810,32 @@ public AllSSTableOpStatus performCleanup(final ColumnFamilyStore cfStore, int jo if (partitioner.getClass() == LocalPartitioner.class) localWrites = RangesAtEndpoint.of(Replica.fullReplica(local, new Range<>(partitioner.getMinimumToken(), partitioner.getMinimumToken()))); - final Set> allRanges = new HashSet<>(localWrites.ranges()); + InUseRanges inUseRanges = getInUseRanges(cfStore.getTableId(), localWrites); + + final List> noLongerOwnedRangesInUseByAccord = inUseRanges.noLongerOwnedRangesInUseByAccord; + final List> allRanges = inUseRanges.allRanges; final Set> transientRanges = new HashSet<>(localWrites.onlyTransient().ranges()); final Set> fullRanges = new HashSet<>(localWrites.onlyFull().ranges()); return parallelAllSSTableOperation(cfStore, new OneSSTableOperation() { + volatile boolean incompleteOperation = false; + @Override public Iterable filterSSTables(LifecycleTransaction transaction) { List sortedSSTables = Lists.newArrayList(transaction.originals()); Iterator sstableIter = sortedSSTables.iterator(); int totalSSTables = 0; - int skippedSStables = 0; + int containedInRangeSkippedSStables = 0; + int inUseByAccordSkippedSStables = 0; while (sstableIter.hasNext()) { SSTableReader sstable = sstableIter.next(); boolean needsCleanupFull = needsCleanup(sstable, fullRanges); boolean needsCleanupTransient = !transientRanges.isEmpty() && sstable.isRepaired() && needsCleanup(sstable, transientRanges); + boolean needsCleanupAccord = needsCleanup(sstable, noLongerOwnedRangesInUseByAccord); + //If there are no ranges for which the table needs cleanup either due to lack of intersection or lack //of the table being repaired. totalSSTables++; @@ -821,11 +850,20 @@ public Iterable filterSSTables(LifecycleTransaction transaction) sstable.isRepaired()); sstableIter.remove(); transaction.cancel(sstable); - skippedSStables++; + containedInRangeSkippedSStables++; + } + else if (!needsCleanupAccord) + { + sstableIter.remove(); + transaction.cancel(sstable); + inUseByAccordSkippedSStables++; + this.incompleteOperation = true; } } - logger.info("Skipping cleanup for {}/{} sstables for {}.{} since they are fully contained in owned ranges (full ranges: {}, transient ranges: {})", - skippedSStables, totalSSTables, cfStore.getKeyspaceName(), cfStore.getTableName(), fullRanges, transientRanges); + + logger.info("Skipping cleanup for {}/{} sstables since they are fully contained in owned ranges and skipping cleanup for {}/{} since they are in use by Accord " + + "for {}.{} (full ranges: {}, transient ranges: {})", containedInRangeSkippedSStables, totalSSTables, inUseByAccordSkippedSStables, totalSSTables, + cfStore.getKeyspaceName(), cfStore.getTableName(), fullRanges, transientRanges); sortedSSTables.sort(SSTableReader.sizeComparator); return sortedSSTables; } @@ -834,11 +872,51 @@ public Iterable filterSSTables(LifecycleTransaction transaction) public void execute(LifecycleTransaction txn) throws IOException { CleanupStrategy cleanupStrategy = CleanupStrategy.get(cfStore, allRanges, transientRanges, txn.onlyOne().isRepaired(), FBUtilities.nowInSeconds()); - doCleanupOne(cfStore, txn, cleanupStrategy, allRanges, hasIndexes); + if (doCleanupOne(cfStore, txn, cleanupStrategy, allRanges, hasIndexes, this.incompleteOperation, noLongerOwnedRangesInUseByAccord)) + this.incompleteOperation = true; + } + + @Override + public boolean incompleteOperation() + { + return this.incompleteOperation; } }, jobs, OperationType.CLEANUP); } + public static class InUseRanges + { + public final List> noLongerOwnedRangesInUseByAccord; + public final List> allRanges; + + public InUseRanges(List> noLongerOwnedRangesInUseByAccord, List> allRanges) + { + this.noLongerOwnedRangesInUseByAccord = noLongerOwnedRangesInUseByAccord; + this.allRanges = allRanges; + } + } + + private InUseRanges getInUseRanges(TableId tableId, RangesAtEndpoint localWrites) + { + List> noLongerOwnedRangesInUseByAccord = new ArrayList<>(); + TableMetadata metadata = Schema.instance.getTableMetadata(tableId); + + if (metadata != null && metadata.requiresAccordSupport()) + { + checkState(localWrites.onlyTransient().ranges().isEmpty(), "Transient Replication is not supported with Accord"); + Map>> inUseRanges = AccordService.instance().getInUseRangesAndMarkRetiredRangesUnsafeToRead(); + if (inUseRanges.containsKey(tableId)) + { + Set> accordOwnedTableRanges = inUseRanges.get(tableId); + for (Range range : accordOwnedTableRanges) + noLongerOwnedRangesInUseByAccord.addAll(range.subtractAll(localWrites.ranges())); + } + } + + Collection> allRanges = Stream.concat(localWrites.ranges().stream(), noLongerOwnedRangesInUseByAccord.stream()).collect(Collectors.toSet()); + return new InUseRanges(Range.normalize(noLongerOwnedRangesInUseByAccord), Range.normalize(allRanges)); + } + public AllSSTableOpStatus performGarbageCollection(final ColumnFamilyStore cfStore, TombstoneOption tombstoneOption, int jobs) throws InterruptedException, ExecutionException { assert !cfStore.isIndex(); @@ -1391,8 +1469,13 @@ public void forceUserDefinedCleanup(String dataFiles) ColumnFamilyStore cfs = entry.getKey(); Keyspace keyspace = cfs.keyspace; final RangesAtEndpoint replicas = StorageService.instance.getLocalReplicas(keyspace.getName()); - final Set> allRanges = replicas.ranges(); + + InUseRanges inUseRanges = getInUseRanges(cfs.getTableId(), replicas); + + final List> noLongerOwnedRangesInUseByAccord = inUseRanges.noLongerOwnedRangesInUseByAccord; + final List> allRanges = inUseRanges.allRanges; final Set> transientRanges = replicas.onlyTransient().ranges(); + boolean hasIndexes = cfs.indexManager.hasIndexes(); SSTableReader sstable = lookupSSTable(cfs, entry.getValue()); @@ -1405,7 +1488,7 @@ public void forceUserDefinedCleanup(String dataFiles) CleanupStrategy cleanupStrategy = CleanupStrategy.get(cfs, allRanges, transientRanges, sstable.isRepaired(), FBUtilities.nowInSeconds()); try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstable, OperationType.CLEANUP)) { - doCleanupOne(cfs, txn, cleanupStrategy, allRanges, hasIndexes); + doCleanupOne(cfs, txn, cleanupStrategy, allRanges, hasIndexes, false, noLongerOwnedRangesInUseByAccord); } catch (IOException e) { @@ -1605,13 +1688,16 @@ public static boolean needsCleanup(SSTableReader sstable, Collection> allRanges, - boolean hasIndexes) throws IOException + private boolean doCleanupOne(final ColumnFamilyStore cfs, + LifecycleTransaction txn, + CleanupStrategy cleanupStrategy, + Collection> allRanges, + boolean hasIndexes, + boolean alreadyIncomplete, + Collection> noLongerOwnedRangesInUseByAccord) throws IOException { assert !cfs.isIndex(); + boolean incompleteOperation = alreadyIncomplete; SSTableReader sstable = txn.onlyOne(); @@ -1621,7 +1707,7 @@ private void doCleanupOne(final ColumnFamilyStore cfs, txn.obsoleteOriginals(); txn.finish(); logger.info("SSTable {} ([{}, {}]) does not intersect the owned ranges ({}), dropping it", sstable, sstable.getFirst().getToken(), sstable.getLast().getToken(), allRanges); - return; + return incompleteOperation; } long start = nanoTime(); @@ -1662,6 +1748,10 @@ private void doCleanupOne(final ColumnFamilyStore cfs, if (notCleaned == null) continue; + if (!noLongerOwnedRangesInUseByAccord.isEmpty() && !incompleteOperation + && Range.isInRanges(partition.partitionKey().getToken(), noLongerOwnedRangesInUseByAccord)) + incompleteOperation = true; + if (writer.append(notCleaned) != null) totalkeysWritten++; @@ -1692,6 +1782,7 @@ private void doCleanupOne(final ColumnFamilyStore cfs, FBUtilities.prettyPrintMemory(endsize), (int) (ratio * 100), totalkeysWritten, dTime)); } + return incompleteOperation; } protected void compactionRateLimiterAcquire(RateLimiter limiter, long bytesScanned, long lastBytesScanned, double compressionRatio) diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 2c5a12ccfc82..1742f5e6668f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -21,8 +21,10 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -87,6 +89,7 @@ import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.SystemKeyspace.BootstrapState; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.RequestExecutionException; import org.apache.cassandra.journal.Params; import org.apache.cassandra.locator.InetAddressAndPort; @@ -1175,6 +1178,25 @@ public void awaitDone(TableId id, long epoch) getBlocking(node.durability().sync("Drop Keyspace/Table (Epoch " + epoch + ')', ExclusiveSyncPoint, TxnId.minForEpoch(epoch), ranges, Self, All, DatabaseDescriptor.getAccordRangeSyncPointTimeoutNanos(), NANOSECONDS), ranges, new LatencyRequestBookkeeping(null), startedAt, deadline, false); } + @Override + public Map>> getInUseRangesAndMarkRetiredRangesUnsafeToRead() + { + Map>> tableToRangeMap = new HashMap<>(); + List globalRanges = getBlocking(node.commandStores().getInUseRangesAndMarkRetiredRangesUnsafeToRead()); + + for (Ranges ranges : globalRanges) + { + for (accord.primitives.Range r : ranges) + { + TokenRange tokenRange = (TokenRange) r; + tableToRangeMap.computeIfAbsent(tokenRange.table(), tableId -> new HashSet<>()) + .add(tokenRange.toKeyspaceRange()); + } + } + + return tableToRangeMap; + } + public Params journalConfiguration() { return journal.configuration(); diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index 4364dd95f7a0..2aaf8e07fc7d 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -19,8 +19,11 @@ package org.apache.cassandra.service.accord; import java.util.Collection; +import java.util.Collections; import java.util.EnumSet; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.BiConsumer; @@ -55,6 +58,8 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.RequestExecutionException; import org.apache.cassandra.journal.Params; import org.apache.cassandra.net.IVerbHandler; @@ -183,6 +188,8 @@ public AccordCompactionInfos(DurableBefore durableBefore, long minEpoch, AccordC void awaitDone(TableId id, long epoch); + Map>> getInUseRangesAndMarkRetiredRangesUnsafeToRead(); + AccordEndpointMapper endpointMapper(); AccordTopologyService topologyService(); @@ -352,6 +359,12 @@ public void awaitDone(TableId id, long epoch) } + @Override + public Map>> getInUseRangesAndMarkRetiredRangesUnsafeToRead() + { + return Collections.emptyMap(); + } + @Override public AccordEndpointMapper endpointMapper() { @@ -562,6 +575,12 @@ public void awaitDone(TableId id, long epoch) delegate.awaitDone(id, epoch); } + @Override + public Map>> getInUseRangesAndMarkRetiredRangesUnsafeToRead() + { + return delegate.getInUseRangesAndMarkRetiredRangesUnsafeToRead(); + } + @Override public Params journalConfiguration() { diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 567df0928342..93c0fdb6cac1 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -480,6 +480,9 @@ private void perform(PrintStream out, String ks, Job job, String jobName) throws case 2: out.printf("Failed marking some sstables compacting in keyspace %s, check server logs for more information.\n", ks); throw new RuntimeException(String.format("Failed marking some sstables compacting in keyspace %s, check server logs for more information.\n", ks)); + case 3: + out.printf("Some SSTables in keyspace %s are still being used by Accord and were not cleaned up, check server logs for more information.\n", ks); + throw new RuntimeException(String.format("Some SSTables in keyspace %s are still being used by Accord and were not cleaned up, check server logs for more information.\n", ks)); } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCleanupTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCleanupTest.java new file mode 100644 index 000000000000..aaeecfb17c99 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCleanupTest.java @@ -0,0 +1,372 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test.accord; + +import java.io.IOException; +import java.util.List; + +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.api.RoutingKey; +import accord.primitives.Ranges; + +import org.apache.cassandra.Util; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.dht.Murmur3Partitioner; +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.NodeToolResult; +import org.apache.cassandra.distributed.api.SimpleQueryResult; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.service.accord.api.TokenKey; + +import static com.google.common.collect.Iterables.getOnlyElement; +import static org.apache.cassandra.service.accord.AccordService.getBlocking; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class AccordCleanupTest extends AccordTestBase +{ + private static final Logger logger = LoggerFactory.getLogger(AccordCleanupTest.class); + + protected String originalToken; + + @Override + protected Logger logger() + { + return logger; + } + + @BeforeClass + public static void setupClass() throws IOException + { + AccordTestBase.setupCluster(builder -> builder + .withoutVNodes() + .appendConfig(config -> config + .set("accord.shard_durability_cycle", "20s") + .with(Feature.GOSSIP, Feature.NETWORK)), 2); + SHARED_CLUSTER.schemaChange("DROP KEYSPACE IF EXISTS " + KEYSPACE + ';'); + SHARED_CLUSTER.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 1}"); + } + + @Before + public void getOriginalToken() + { + originalToken = SHARED_CLUSTER.get(1).callOnInstance(() -> getOnlyElement(StorageService.instance.getTokens())); + } + + @After + public void reset() + { + String token = originalToken; + SHARED_CLUSTER.get(1).runOnInstance(() -> { + StorageService.instance.move(token); + }); + } + + @Test + public void accordNodetoolCleanupTest() throws Throwable + { + String tableName = "tbl0"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'", cluster -> { + setupForCleanupTest(cluster, tableName, qualifiedTableName); + + NodeToolResult nodetoolResult = cluster.get(1).nodetoolResult("cleanup", KEYSPACE, tableName); + + assertEquals(0, nodetoolResult.getRc()); + assertEquals(0, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + // Ensure data is cleaned up + assertEquals(0, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + }); + } + + @Test + public void accordUserDefinedCleanupTest() throws Throwable + { + String tableName = "tbl1"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'", cluster -> { + setupForCleanupTest(cluster, tableName, qualifiedTableName); + + cluster.get(1).runOnInstance(() -> { + String dataFile = getOnlyElement(Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables()).getFilename(); + CompactionManager.instance.forceUserDefinedCleanup(dataFile); + }); + + // Ensure data is cleaned up + assertEquals(0, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + }); + } + + @Test + public void accordNodetoolCleanupPartialSSTableTest() throws Throwable + { + String tableName = "tbl2"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'", cluster -> { + setupForCleanupPartialSSTableTest(cluster, tableName, qualifiedTableName); + + NodeToolResult nodetoolResult = cluster.get(1).nodetoolResult("cleanup", KEYSPACE, tableName); + + assertEquals(0, nodetoolResult.getRc()); + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + // Ensure data is correct + assertEquals(1, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + assertEquals(0, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 2 LIMIT 1").length); + }); + } + + @Test + public void accordUserDefinedCleanupPartialSSTableTest() throws Throwable + { + String tableName = "tbl3"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'", cluster -> { + setupForCleanupPartialSSTableTest(cluster, tableName, qualifiedTableName); + + cluster.get(1).runOnInstance(() -> { + String dataFile = getOnlyElement(Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables()).getFilename(); + CompactionManager.instance.forceUserDefinedCleanup(dataFile); + }); + + // Ensure data is correct + assertEquals(1, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + assertEquals(0, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 2 LIMIT 1").length); + }); + } + + @Test + public void accordNodetoolCleanupRangeInUseTest() throws Throwable + { + String tableName = "tbl4"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'", cluster -> { + setupForCleanupRangeInUseTest(cluster, tableName, qualifiedTableName); + + NodeToolResult nodetoolResult = cluster.get(1).nodetoolResult("cleanup", KEYSPACE, tableName); + + assertTrue(nodetoolResult.getStdout().contains("Some SSTables in keyspace " + KEYSPACE + " are still being used by Accord and were not cleaned up, check server logs for more information.")); + assertEquals(2, nodetoolResult.getRc()); + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + // Ensure data is still there + assertEquals(1, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + }); + } + + @Test + public void accordUserDefinedCleanupRangeInUseTest() throws Throwable + { + String tableName = "tbl5"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'", cluster -> { + setupForCleanupRangeInUseTest(cluster, tableName, qualifiedTableName); + + cluster.get(1).runOnInstance(() -> { + String dataFile = getOnlyElement(Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables()).getFilename(); + CompactionManager.instance.forceUserDefinedCleanup(dataFile); + }); + + // Ensure data is still there + assertEquals(1, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + }); + } + + @Test + public void nodetoolCleanupForNonAccordTableTest() throws Throwable + { + String tableName = "tbl6"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int)", cluster -> { + setupForCleanupForNonAccordTableTest(cluster, tableName, qualifiedTableName); + + NodeToolResult nodetoolResult = cluster.get(1).nodetoolResult("cleanup", KEYSPACE, tableName); + + assertEquals(0, nodetoolResult.getRc()); + assertEquals(0, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + // Ensure data is cleaned up + assertEquals(0, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + }); + } + + @Test + public void userDefinedCleanupForNonAccordTableTest() throws Throwable + { + String tableName = "tbl7"; + String qualifiedTableName = KEYSPACE + '.' + tableName; + + test("CREATE TABLE " + qualifiedTableName + " (k int PRIMARY KEY, v int)", cluster -> { + setupForCleanupForNonAccordTableTest(cluster, tableName, qualifiedTableName); + + cluster.get(1).runOnInstance(() -> { + String dataFile = getOnlyElement(Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables()).getFilename(); + CompactionManager.instance.forceUserDefinedCleanup(dataFile); + }); + + // Ensure data is cleaned up + assertEquals(0, cluster.get(1).executeInternal("SELECT k FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1").length); + }); + } + + private void setupForCleanupTest(Cluster cluster, String tableName, String qualifiedTableName) + { + cluster.coordinator(1).execute(wrapInTxn("INSERT INTO " + qualifiedTableName + " (k, v) VALUES (?, ?)"), ConsistencyLevel.SERIAL, 1, 2); + SimpleQueryResult result = cluster.coordinator(1).executeWithResult("SELECT token(k) FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1", ConsistencyLevel.SERIAL); + + cluster.get(1).flush(withKeyspace("%s")); + + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + long token = (Long) result.toObjectArrays()[0][0]; + + assertTrue(token < Long.parseLong(originalToken)); + + // Cluster 1 no longer owns token + cluster.get(1).runOnInstance(() -> { + AccordService.instance().node().durability().shards().start(); + StorageService.instance.move(Long.toString(token - 1000)); + }); + + // Wait until Accord retires range, so it no longer has ownership of token + cluster.get(1).runOnInstance(() -> { + TableId tid = Schema.instance.getTableMetadata(KEYSPACE, tableName).id(); + RoutingKey key = TokenKey.parse(tid, String.valueOf(token), Murmur3Partitioner.instance); + + Util.spinUntilTrue(() -> { + boolean doesNotContainsToken = true; + List inUseRanges = getBlocking(AccordService.instance().node().commandStores().getInUseRangesAndMarkRetiredRangesUnsafeToRead()); + for (Ranges ranges : inUseRanges) + { + if (ranges.intersects(key)) + doesNotContainsToken = false; + } + return doesNotContainsToken; + }, 30); + }); + } + + private void setupForCleanupPartialSSTableTest(Cluster cluster, String tableName, String qualifiedTableName) + { + cluster.coordinator(1).execute(wrapInTxn("INSERT INTO " + qualifiedTableName + " (k, v) VALUES (?, ?)"), ConsistencyLevel.SERIAL, 1, 2); + cluster.coordinator(1).execute(wrapInTxn("INSERT INTO " + qualifiedTableName + " (k, v) VALUES (?, ?)"), ConsistencyLevel.SERIAL, 2, 2); + + SimpleQueryResult result1 = cluster.coordinator(1).executeWithResult("SELECT token(k) FROM " + qualifiedTableName + " WHERE k = 2 LIMIT 1", ConsistencyLevel.SERIAL); + SimpleQueryResult result2 = cluster.coordinator(1).executeWithResult("SELECT token(k) FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1", ConsistencyLevel.SERIAL); + + cluster.get(1).flush(withKeyspace("%s")); + + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + long token1 = (Long) result1.toObjectArrays()[0][0]; + long token2 = (Long) result2.toObjectArrays()[0][0]; + long newToken = token1 - 1000; + + assertTrue((token2 < newToken) && (token1 < Long.parseLong(originalToken))); + + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + // Cluster 1 now only owns token2, but Accord still requires token1 + cluster.get(1).runOnInstance(() -> { + AccordService.instance().node().durability().shards().start(); + StorageService.instance.move(Long.toString(newToken)); + }); + + // Wait until Accord retires range, so it no longer has ownership of token1 + cluster.get(1).runOnInstance(() -> { + TableId tid = Schema.instance.getTableMetadata(KEYSPACE, tableName).id(); + RoutingKey key = TokenKey.parse(tid, String.valueOf(token1), Murmur3Partitioner.instance); + + Util.spinUntilTrue(() -> { + boolean doesNotContainsToken = true; + List inUseRanges = getBlocking(AccordService.instance().node().commandStores().getInUseRangesAndMarkRetiredRangesUnsafeToRead()); + for (Ranges ranges : inUseRanges) + { + if (ranges.intersects(key)) + doesNotContainsToken = false; + } + return doesNotContainsToken; + }, 30); + }); + } + + private void setupForCleanupRangeInUseTest(Cluster cluster, String tableName, String qualifiedTableName) + { + cluster.coordinator(1).execute(wrapInTxn("INSERT INTO " + qualifiedTableName + " (k, v) VALUES (?, ?)"), ConsistencyLevel.SERIAL, 1, 2); + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult("SELECT token(k) FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1", ConsistencyLevel.SERIAL); + + cluster.get(1).flush(withKeyspace("%s")); + + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + long token = (Long) result.toObjectArrays()[0][0]; + + assertTrue(token < Long.parseLong(originalToken)); + + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + cluster.get(1).runOnInstance(() -> { + AccordService.instance().node().durability().shards().stop(); + StorageService.instance.move(Long.toString(token - 1000)); + }); + } + + private void setupForCleanupForNonAccordTableTest(Cluster cluster, String tableName, String qualifiedTableName) + { + cluster.coordinator(1).execute("INSERT INTO " + qualifiedTableName + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, 1, 2); + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult("SELECT token(k) FROM " + qualifiedTableName + " WHERE k = 1 LIMIT 1", ConsistencyLevel.SERIAL); + + cluster.get(1).flush(withKeyspace("%s")); + + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + long token = (Long) result.toObjectArrays()[0][0]; + + assertTrue(token < Long.parseLong(originalToken)); + + assertEquals(1, (int) cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName).getLiveSSTables().size())); + + cluster.get(1).runOnInstance(() -> { + AccordService.instance().node().durability().shards().stop(); + StorageService.instance.move(Long.toString(token - 1000)); + }); + } +}