From 964f88a7ef2a237c0b03349d0b77f323e3c47076 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Thu, 30 Jul 2026 15:27:53 -0700 Subject: [PATCH] fixes --- .../cassandra/service/accord/txn/TxnRead.java | 32 +- .../test/accord/AccordImportSSTableTest.java | 290 ++++++------------ 2 files changed, 102 insertions(+), 220 deletions(-) diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java index 8e09b88d80..e27cdc20e7 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java @@ -43,7 +43,6 @@ import accord.primitives.Routable.Domain; import accord.primitives.Seekable; import accord.primitives.Seekables; import accord.primitives.Timestamp; -import accord.utils.Invariants; import accord.utils.UnhandledEnum; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; @@ -88,8 +87,8 @@ import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSi public class TxnRead extends AbstractKeySorted implements Read { - private static final TxnRead EMPTY_KEY = new TxnRead(TableMetadatas.none(), Domain.Key); - private static final TxnRead EMPTY_RANGE = new TxnRead(TableMetadatas.none(), Domain.Range); + private static final TxnRead EMPTY_KEY = new TxnRead(TableMetadatas.none(), Domain.Key, null); + private static final TxnRead EMPTY_RANGE = new TxnRead(TableMetadatas.none(), Domain.Range, null); private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY_KEY); private static final Comparator TXN_NAMED_READ_KEY_COMPARATOR = Comparator.comparing(a -> ((PartitionKey) a.key())); private static final byte TYPE_EMPTY_KEY = 0; @@ -121,13 +120,13 @@ public class TxnRead extends AbstractKeySorted implements Read @Nullable private final ImportMetadata importMetadata; - private TxnRead(TableMetadatas tables, Domain domain) + private TxnRead(TableMetadatas tables, Domain domain, @Nullable ImportMetadata importMetadata) { super(new TxnNamedRead[0], domain); this.tables = tables; this.domain = domain; this.cassandraConsistencyLevel = null; - this.importMetadata = null; + this.importMetadata = importMetadata; } private TxnRead(TableMetadatas tables, @Nonnull TxnNamedRead[] items, @Nullable ConsistencyLevel cassandraConsistencyLevel, @Nullable ImportMetadata importMetadata) @@ -169,6 +168,14 @@ public class TxnRead extends AbstractKeySorted implements Read return new TxnRead(tables, items, consistencyLevel, null); } + public static TxnRead createTxnRead(TableMetadatas tables, @Nonnull List items, @Nullable ConsistencyLevel consistencyLevel, Domain domain, @Nullable ImportMetadata importMetadata) + { + if (items.isEmpty()) + return new TxnRead(TableMetadatas.none(), domain, importMetadata); + sortReads(items); + return new TxnRead(tables, items, consistencyLevel, importMetadata); + } + public static TxnRead createSerialRead(List readCommands, ConsistencyLevel consistencyLevel, TableMetadatasAndKeys.KeyCollector keyCollector) { List reads = new ArrayList<>(readCommands.size()); @@ -316,12 +323,7 @@ public class TxnRead extends AbstractKeySorted implements Read throw new UnhandledEnum(select.domain()); } - if (isUsedForImport()) - { - Invariants.require(!reads.isEmpty()); - return new TxnRead(tables, reads, cassandraConsistencyLevel, importMetadata); - } - return createTxnRead(tables, reads, cassandraConsistencyLevel, select.domain()); + return createTxnRead(tables, reads, cassandraConsistencyLevel, select.domain(), importMetadata); } @Override @@ -402,12 +404,8 @@ public class TxnRead extends AbstractKeySorted implements Read break; } } - if (read.isUsedForImport()) - { - Invariants.require(!reads.isEmpty()); - return new TxnRead(tables, reads, cassandraConsistencyLevel, importMetadata); - } - return createTxnRead(tables, reads, cassandraConsistencyLevel, that.domain); + + return createTxnRead(tables, reads, cassandraConsistencyLevel, that.domain, importMetadata); } public void unmemoize() 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 067ddeb966..5d7808210f 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java @@ -67,7 +67,6 @@ 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 @@ -80,29 +79,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testImportSSTables() throws Throwable { - String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); - - CQLSSTableWriter.Builder builder1 = CQLSSTableWriter.builder() - .forTable(TABLE_SCHEMA_CQL) - .inDirectory(file) - .using("INSERT INTO " + KEYSPACE_TABLE + "(k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder1.build()) - { - writer.addRow(1, 1); - writer.addRow(2, 1); - } - - CQLSSTableWriter.Builder builder2 = CQLSSTableWriter.builder() - .forTable(TABLE_SCHEMA_CQL) - .inDirectory(file) - .using("INSERT INTO " + KEYSPACE_TABLE + "(k, v) " + "VALUES (?, ?)"); - - - try (CQLSSTableWriter writer = builder2.build()) - { - writer.addRow(3, 1); - } + String file = writeSSTables(new int[] { 1, 2 }, new int[] { 3 }); try (Cluster cluster = init(builder().withNodes(3) .withoutVNodes() @@ -111,16 +88,7 @@ public class AccordImportSSTableTest extends TestBaseImpl 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'"); - - // Disable autocompaction so when we go to check the number of SSTables they correspond to the SSTables that we have imported - cluster.forEach(instance -> { - instance.runOnInstance(() -> { - ColumnFamilyStore.getIfExists(KEYSPACE, TABLE).disableAutoCompaction(); - }); - }); + createSchema(cluster); cluster.get(1).runOnInstance(() -> { ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); @@ -146,17 +114,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testImportSSTablesBuildsIndex() throws Throwable { - String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); - - CQLSSTableWriter.Builder builder1 = CQLSSTableWriter.builder() - .forTable(TABLE_SCHEMA_CQL) - .inDirectory(file) - .using("INSERT INTO " + KEYSPACE_TABLE + "(k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder1.build()) - { - writer.addRow(1, 1); - } + String file = writeSSTables(new int[] { 1 }); try (Cluster cluster = init(builder().withNodes(3) .withoutVNodes() @@ -165,9 +123,7 @@ public class AccordImportSSTableTest extends TestBaseImpl 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'"); + createSchema(cluster); String indexName = "v_idx"; cluster.schemaChange(withKeyspace("CREATE INDEX " + indexName + " ON " + KEYSPACE_TABLE + " (v) USING 'sai'")); @@ -197,17 +153,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testSSTableImportWithConcurrentTopologyChangeFails() 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_TABLE + "(k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder.build()) - { - writer.addRow(1, 1); - } + String file = writeSSTables(new int[] { 1 }); try (Cluster cluster = init(builder().withNodes(3) .withoutVNodes() @@ -217,9 +163,7 @@ public class AccordImportSSTableTest extends TestBaseImpl 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'"); + createSchema(cluster); Thread importer = new Thread(() -> { cluster.get(1).runOnInstance(() -> { @@ -254,28 +198,14 @@ public class AccordImportSSTableTest extends TestBaseImpl @Ignore public void testSSTableImportReplicaDown() 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_TABLE + "(k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder.build()) - { - writer.addRow(1, 1); - writer.addRow(2, 1); - writer.addRow(3, 1); - } + String file = writeSSTables(new int[] { 1, 2, 3 }); int FAILED_REPLICA = 2; try (Cluster cluster = init(builder().withNodes(3).withoutVNodes() .withDataDirCount(1).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'"); + createSchema(cluster); cluster.get(FAILED_REPLICA).shutdown().get(); @@ -301,19 +231,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testSSTableImportStreamingFailedCleanup() 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_TABLE + " (k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder.build()) - { - writer.addRow(1, 1); - writer.addRow(2, 1); - writer.addRow(3, 1); - } + String file = writeSSTables(new int[] { 1, 2, 3 }); int FAILED_STREAM = 3; try (Cluster cluster = init(builder().withNodes(3) @@ -324,9 +242,7 @@ public class AccordImportSSTableTest extends TestBaseImpl 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'"); + createSchema(cluster); cluster.get(1).runOnInstance(() -> { ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); @@ -347,19 +263,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testSSTableImportBounceAfterPending() 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_TABLE + " (k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder.build()) - { - writer.addRow(1, 1); - writer.addRow(2, 1); - writer.addRow(3, 1); - } + String file = writeSSTables(new int[] { 1, 2, 3 }); // We disable local delivery so that we can stimulate a network partition by dropping // ACCORD_STABLE_THEN_READ_REQ messages @@ -369,9 +273,7 @@ public class AccordImportSSTableTest extends TestBaseImpl .set("accord.permit_local_delivery", false) .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'"); + createSchema(cluster); // Prevent SSTables from being moved to the live set cluster.filters().outbound().verbs(Verb.ACCORD_STABLE_THEN_READ_REQ.id).drop(); @@ -393,19 +295,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testRecoveryCoordinatorPerformsImport() 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_TABLE + " (k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder.build()) - { - writer.addRow(1, 1); - writer.addRow(2, 1); - writer.addRow(3, 1); - } + String file = writeSSTables(new int[] { 1, 2, 3 }); // We disable local delivery so that we can stimulate a network partition by dropping // ACCORD_STABLE_THEN_READ_REQ messages @@ -416,9 +306,7 @@ public class AccordImportSSTableTest extends TestBaseImpl .set("accord.permit_local_delivery", false) .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'"); + createSchema(cluster); // Simulate a network partition right before the StableThenRead message is sent // so that a recovery coordinator can pick up the ImportTxn @@ -443,16 +331,9 @@ public class AccordImportSSTableTest extends TestBaseImpl .hasMessageContaining("SSTable import failed locally; however the operation may still be applied by the recovery coordinator"); }); - // Wait until the recovery coordinator picks up the Import Txn Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS); - for (int i = 1; i <= 3; i++) - { - cluster.get(i).runOnInstance(() -> { - ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); - assertEquals(1, cfs.getLiveSSTables().size()); - }); - } + assertSSTableCount(cluster, 1); assertLocalSelect(cluster, rows -> { assertRows(rows, row(1, 1), row(2, 1), row(3, 1)); }); } @@ -461,19 +342,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testRecoveryCoordinatorPerformsImport2() 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_TABLE + " (k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder.build()) - { - writer.addRow(1, 1); - writer.addRow(2, 1); - writer.addRow(3, 1); - } + String file = writeSSTables(new int[] { 1, 2, 3 }); try (Cluster cluster = init(builder().withNodes(3).withoutVNodes() .withDataDirCount(1).withConfig((config) -> @@ -482,9 +351,7 @@ public class AccordImportSSTableTest extends TestBaseImpl .set("accord.permit_local_delivery", false) .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'"); + createSchema(cluster); // Node 1 sends the StableThenRead message to node 2 and then node 1 fails, so the // only existence of the stable message is at node 2 @@ -521,7 +388,6 @@ public class AccordImportSSTableTest extends TestBaseImpl .hasMessageContaining("SSTable import failed locally; however the operation may still be applied by the recovery coordinator"); }); - // Wait until the recovery coordinator picks up the Import Txn Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS); Iterable up = cluster.stream() @@ -537,29 +403,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @Test public void testImportSSTablesCleanupWithMultipleDataDirectories() throws Throwable { - String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); - - CQLSSTableWriter.Builder builder1 = CQLSSTableWriter.builder() - .forTable(TABLE_SCHEMA_CQL) - .inDirectory(file) - .using("INSERT INTO " + KEYSPACE_TABLE + "(k, v) " + "VALUES (?, ?)"); - - try (CQLSSTableWriter writer = builder1.build()) - { - writer.addRow(1, 1); - writer.addRow(2, 1); - } - - CQLSSTableWriter.Builder builder2 = CQLSSTableWriter.builder() - .forTable(TABLE_SCHEMA_CQL) - .inDirectory(file) - .using("INSERT INTO " + KEYSPACE_TABLE + "(k, v) " + "VALUES (?, ?)"); - - - try (CQLSSTableWriter writer = builder2.build()) - { - writer.addRow(3, 1); - } + String file = writeSSTables(new int[] { 1, 2 }, new int[] { 3 }); try (Cluster cluster = init(builder().withNodes(3) .withoutVNodes() @@ -568,16 +412,7 @@ public class AccordImportSSTableTest extends TestBaseImpl 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'"); - - // Disable autocompaction so when we go to check the number of SSTables they correspond to the SSTables that we have imported - cluster.forEach(instance -> { - instance.runOnInstance(() -> { - ColumnFamilyStore.getIfExists(KEYSPACE, TABLE).disableAutoCompaction(); - }); - }); + createSchema(cluster); cluster.get(1).runOnInstance(() -> { ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); @@ -587,14 +422,10 @@ public class AccordImportSSTableTest extends TestBaseImpl Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS); - // Assert that each node has 2 SSTables assertSSTableCount(cluster, 2); - // Assert that each node has the correct values assertLocalSelect(cluster, rows -> { assertRows(rows, row(1, 1), row(2, 1), row(3, 1)); }); - - // Assert that SSTables are moved from the pending directories assertPendingDirs(cluster, (File pendingUuidDir) -> { Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty(); }); @@ -604,18 +435,7 @@ public class AccordImportSSTableTest extends TestBaseImpl @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_TABLE + " (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); - } + String file = writeSSTables(new int[] { 1 }); try (Cluster cluster = init(builder().withNodes(3).withoutVNodes() .withDataDirCount(1) @@ -624,9 +444,7 @@ public class AccordImportSSTableTest extends TestBaseImpl 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'"); + createSchema(cluster); cluster.get(1).runOnInstance(() -> { ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); @@ -642,6 +460,72 @@ public class AccordImportSSTableTest extends TestBaseImpl } } + @Test + public void testImportAppliesOnPartitionedReplica() throws Throwable + { + String file = writeSSTables(new int[] { 1, 2, 3 }); + + try (Cluster cluster = init(builder().withNodes(3) + .withoutVNodes() + .withDataDirCount(1) + .withConfig(config -> config.with(Feature.NETWORK, Feature.GOSSIP)) + .start())) + { + createSchema(cluster); + + // Partition node3 from the consensus phases only. + cluster.filters() + .verbs(Verb.ACCORD_PRE_ACCEPT_REQ.id, + Verb.ACCORD_ACCEPT_REQ.id, + Verb.ACCORD_NOT_ACCEPT_REQ.id, + Verb.ACCORD_COMMIT_REQ.id, + Verb.ACCORD_STABLE_THEN_READ_REQ.id, + Verb.ACCORD_READ_REQ.id) + .to(3) + .drop(); + + cluster.get(1).runOnInstance(() -> { + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); + cfs.importNewSSTables(Set.of(file), true, true, true, true, true, true, true); + }); + + Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS); + + assertLocalSelect(cluster, rows -> assertRows(rows, row(1, 1), row(2, 1), row(3, 1))); + assertSSTableCount(cluster, 1); + } + } + + private static void createSchema(Cluster cluster) + { + 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'"); + + // Disable autocompaction so when we go to check the number of SSTables they correspond to the SSTables that we have imported + cluster.forEach(instance -> instance.runOnInstance(() -> ColumnFamilyStore.getIfExists(KEYSPACE, TABLE).disableAutoCompaction())); + } + + private static String writeSSTables(int[]... sstables) throws Exception + { + String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + for (int[] sstable : sstables) + { + CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE_TABLE + " (k, v) " + "VALUES (?, ?)"); + + try (CQLSSTableWriter writer = builder.build()) + { + for (int key : sstable) + writer.addRow(key, 1); + } + } + + return file; + } + private static void bounce(Cluster cluster) { cluster.forEach(instance -> {