diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java index b92cef2346..e4fb72b1d2 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java @@ -24,6 +24,7 @@ import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; import com.google.common.util.concurrent.Uninterruptibles; @@ -43,10 +44,14 @@ import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.api.IIsolatedExecutor; import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.io.sstable.CQLSSTableWriter; +import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.io.sstable.Component; +import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.File; import org.apache.cassandra.net.Verb; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.CoordinatedTransfer; +import org.apache.cassandra.service.accord.LocalTransfers; import org.apache.cassandra.utils.Shared; import org.assertj.core.api.Assertions; @@ -55,10 +60,12 @@ import org.junit.Test; import static com.google.common.collect.Iterables.getOnlyElement; import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; import static net.bytebuddy.matcher.ElementMatchers.takesNoArguments; import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; import static org.apache.cassandra.distributed.shared.AssertUtils.row; import static org.junit.Assert.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; public class AccordImportSSTableTest extends TestBaseImpl { @@ -436,7 +443,7 @@ public class AccordImportSSTableTest extends TestBaseImpl Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS); - // Assert that each node has 2 SSTables + // Assert that each node has 1 SSTable cluster.forEach(instance -> { instance.runOnInstance(() -> { ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, "tbl"); @@ -454,6 +461,62 @@ public class AccordImportSSTableTest extends TestBaseImpl } } + @Test + public void testWithZeroCopyStreaming() throws Throwable + { + + } + + /** + * There is a bug with this case, because we are going to be marked as stable and then perform the read which has the + * TxnImport logic contained within it. For ImportTxn's, we need to retry the read. + */ + @Test + public void testImportSSTableFailsActivation() throws Throwable + { + String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + + CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + // We import a 1 token SSTable so the txn read logic is not run multiple of times by different CommandStores + try (CQLSSTableWriter writer = builder.build()) + { + writer.addRow(1, 1); + } + + try (Cluster cluster = init(builder().withNodes(3).withoutVNodes() + .withDataDirCount(1) + .withInstanceInitializer(ByteBuddyInjections.FailDiskMove.install(2)) + .withConfig((config) -> + config + .with(Feature.NETWORK, Feature.GOSSIP)).start())) + { + cluster.schemaChange("DROP KEYSPACE IF EXISTS " + KEYSPACE); + cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}"); + cluster.schemaChange("CREATE TABLE " + KEYSPACE_TABLE + " (k int PRIMARY KEY, v int) WITH transactional_mode='full'"); + + cluster.get(1).runOnInstance(() -> { + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); + Set paths = Set.of(file); + cfs.importNewSSTables(paths, true, true, true, true, true, true, true); + }); + + Uninterruptibles.sleepUninterruptibly(30, TimeUnit.SECONDS); + + cluster.forEach(instance -> { + instance.runOnInstance(() -> { + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, "tbl"); + assertEquals(1, cfs.getLiveSSTables().size()); + }); + }); + + assertLocalSelect(cluster, rows -> { assertRows(rows, row(1, 1), row(2, 1), row(3, 1)); }); + } + } + private static void bounce(Cluster cluster) { cluster.forEach(instance -> { @@ -469,6 +532,17 @@ public class AccordImportSSTableTest extends TestBaseImpl }); } + private static void assertLocalTransferIsCleanup(Iterable validate) + { + for (IInvokableInstance instance : validate) + { + instance.runOnInstance(() -> { + assertTrue(LocalTransfers.instance.local.isEmpty()); + assertTrue(LocalTransfers.instance.coordinating.isEmpty()); + }); + } + } + private static void assertPendingDirs(Iterable validate, IIsolatedExecutor.SerializableConsumer forPendingUuidDir) { for (IInvokableInstance instance : validate) @@ -502,6 +576,7 @@ public class AccordImportSSTableTest extends TestBaseImpl public static class State { public static CountDownLatch waitForTopologyChange = new CountDownLatch(1); + public static AtomicBoolean shouldFailDisk = new AtomicBoolean(true); } public static class ByteBuddyInjections @@ -550,5 +625,32 @@ public class AccordImportSSTableTest extends TestBaseImpl throw new RuntimeException("Failing incoming stream for test"); } } + + public static class FailDiskMove + { + public static IInstanceInitializer install(int... nodes) + { + return (ClassLoader cl, ThreadGroup tg, int num, int generation) -> { + for (int node : nodes) + if (node == num) + new ByteBuddy().rebase(SSTableReader.class) + .method(named("moveAndOpenSSTable").and(takesArguments(5))) + .intercept(MethodDelegation.to(FailDiskMove.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + }; + } + + @SuppressWarnings("unused") + public static SSTableReader moveAndOpenSSTable(ColumnFamilyStore cfs, Descriptor oldDescriptor, Descriptor newDescriptor, Set components, boolean copyData, @SuperCall Callable r) throws Exception + { + if (State.shouldFailDisk.get()) + { + State.shouldFailDisk.set(false); + throw new RuntimeException("Failing move and open SSTable"); + } + return r.call(); + } + } } }