diff --git a/src/java/org/apache/cassandra/db/SSTableImporter.java b/src/java/org/apache/cassandra/db/SSTableImporter.java index 2c9435e72e..2ae728a776 100644 --- a/src/java/org/apache/cassandra/db/SSTableImporter.java +++ b/src/java/org/apache/cassandra/db/SSTableImporter.java @@ -247,8 +247,16 @@ public class SSTableImporter } catch (Throwable t) { - logger.error("[{}] Failed adding SSTables", importID, t); - throw new RuntimeException("Failed adding SSTables", t); + if (isAccordEnabled) + { + String msg = "Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator"; + throw new RuntimeException(msg, t); + } + else + { + logger.error("[{}] Failed adding SSTables", importID, t); + throw new RuntimeException("Failed adding SSTables", t); + } } logger.info("[{}] Done loading load new SSTables for {}/{}", importID, cfs.getKeyspaceName(), cfs.getTableName()); diff --git a/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java b/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java index f70ebad78e..71a6eb7a84 100644 --- a/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java +++ b/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java @@ -50,6 +50,7 @@ import org.apache.cassandra.db.streaming.CassandraOutgoingFile; import org.apache.cassandra.dht.NormalizedRanges; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.ReadTimeoutException; import org.apache.cassandra.exceptions.RequestFailure; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.locator.InetAddressAndPort; @@ -96,6 +97,8 @@ public class CoordinatedTransfer final Map nodeStreamingContext; SingleTransferResult streamResult = SingleTransferResult.Init(); + // If importTxnEpochMismatch is true, all replicas will deterministically not import the SSTables in the pending directory + volatile boolean importTxnEpochMismatch = false; public CoordinatedTransfer(UUID importID, TableMetadata tableMetadata, Map nodeStreamingContext, long streamingEpoch, TokenRange allSSTableRanges) { @@ -117,14 +120,27 @@ public class CoordinatedTransfer LocalTransfers.instance().save(this); stream(); - performImportTxn(); - LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this); + + try + { + performImportTxn(); + if (importTxnEpochMismatch) + { + LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this); + throw new RuntimeException("SSTable import failed because of a concurrent topology change; please retry the operation"); + } + LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this); + } + catch (ReadTimeoutException e) + { + throw new RuntimeException("SSTable import failed locally; however the operation may still be applied by the recovery coordinator", e); + } } private void performImportTxn() { TableMetadatas tables = TableMetadatas.of(tableMetadata); - TxnRead read = TxnRead.createImport(tables, allSSTableRanges, streamResult.planId, streamingEpoch); + TxnRead read = TxnRead.createImport(tables, allSSTableRanges, importID, streamResult.planId, streamingEpoch); TableMetadatasAndKeys tablesAndKeys = new TableMetadatasAndKeys(tables, read.keys()); Txn txn = new Txn.InMemory(Read, read.keys(), read, TxnQuery.NONE, null, tablesAndKeys); IAccordService.IAccordResult accordResult = AccordService.instance().coordinateAsync(tableMetadata.epoch.getEpoch(), txn, ConsistencyLevel.ALL, Dispatcher.RequestTime.forImmediateExecution()); diff --git a/src/java/org/apache/cassandra/service/accord/LocalTransfers.java b/src/java/org/apache/cassandra/service/accord/LocalTransfers.java index 371bd3a993..d6390e0345 100644 --- a/src/java/org/apache/cassandra/service/accord/LocalTransfers.java +++ b/src/java/org/apache/cassandra/service/accord/LocalTransfers.java @@ -181,16 +181,14 @@ public class LocalTransfers }); } - // This method will be called by every Accord command store executor of - // the ranges it intersects - public void activatePendingTransfers(TxnRead.ImportMetadata metadata) + public void activatePendingTransfers(TxnRead.ImportMetadata metadata, long executeAtEpoch) { lock.readLock().lock(); try { PendingLocalTransfer pendingLocalTransfer = local.get(metadata.getPlanId()); if (pendingLocalTransfer != null) - pendingLocalTransfer.activate(); + pendingLocalTransfer.activate(metadata, executeAtEpoch); } finally { diff --git a/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java b/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java index 71c96c8fa9..2520814a19 100644 --- a/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java +++ b/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java @@ -33,6 +33,7 @@ import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.File; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.accord.txn.TxnRead; import org.apache.cassandra.utils.TimeUUID; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; @@ -66,11 +67,31 @@ public class PendingLocalTransfer /** * Safely moves SSTables into the live set. */ - public synchronized void activate() + public synchronized void activate(TxnRead.ImportMetadata metadata, long executeAtEpoch) { if (activated) return; + CoordinatedTransfer coordinatedTransfer = LocalTransfers.instance.coordinating.get(metadata.getImportID()); + boolean isCoordinator = coordinatedTransfer != null; + + if (metadata.getStreamingEpoch() != executeAtEpoch) + { + logger.info("{} Failing activation of pending SSTables because streaming epoch {} != importTxn executeAt epoch {}", + logPrefix(), metadata.getStreamingEpoch(), executeAtEpoch); + + if (isCoordinator) + { + coordinatedTransfer.importTxnEpochMismatch = true; + LocalTransfers.instance().scheduleCoordinatedTransferCleanup(coordinatedTransfer); + } + else + LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId); + + activated = true; + return; + } + long startedActivation = currentTimeMillis(); logger.info("{} Activating transfer {}, {} ms since pending", logPrefix(), this, startedActivation - createdAt); ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(tableId); @@ -117,7 +138,10 @@ public class PendingLocalTransfer long finishedActivation = currentTimeMillis(); logger.info("{} Finished activating transfer {} in {} ms", logPrefix(), this, finishedActivation - startedActivation); - LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId); + if (isCoordinator) + LocalTransfers.instance().scheduleCoordinatedTransferCleanup(coordinatedTransfer); + else + LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java index c2f871df86..5905fc1b24 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java @@ -413,9 +413,7 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized performSSTableImport(AccordExecutor executor, TxnRead.ImportMetadata importMetadata, Timestamp executeAt) { Callable callable = () -> { - if (importMetadata.getStreamingEpoch() != executeAt.epoch()) - throw new RuntimeException("SSTable import failed because of a concurrent topology change"); - LocalTransfers.instance.activatePendingTransfers(importMetadata); + LocalTransfers.instance.activatePendingTransfers(importMetadata, executeAt.epoch()); return new TxnData(); }; return submit(executor, callable, callable); 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 f9567c07cf..8e09b88d80 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java @@ -23,6 +23,7 @@ import java.util.ArrayList; import java.util.Comparator; import java.util.List; import java.util.Objects; +import java.util.UUID; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -67,6 +68,7 @@ import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys; import org.apache.cassandra.service.accord.serializers.Version; import org.apache.cassandra.utils.ObjectSizes; import org.apache.cassandra.utils.TimeUUID; +import org.apache.cassandra.utils.UUIDSerializer; import static accord.primitives.Routables.Slice.Minimal; import static accord.utils.Invariants.require; @@ -204,9 +206,9 @@ public class TxnRead extends AbstractKeySorted implements Read // the ImportTxn has custom logic to move SSTables in the pending directory to the live set. Because these txn's are performed // using range reads, clients can see a window of inconsistency for read txn's that are concurrent with the ImportTxn. However, // data will always be consistent. - public static TxnRead createImport(TableMetadatas tables, TokenRange range, TimeUUID planId, long streamingEpoch) + public static TxnRead createImport(TableMetadatas tables, TokenRange range, UUID importID, TimeUUID planId, long streamingEpoch) { - ImportMetadata importMetadata = new ImportMetadata(planId, streamingEpoch); + ImportMetadata importMetadata = new ImportMetadata(importID, planId, streamingEpoch); return new TxnRead(tables, ImmutableList.of(new TxnNamedRead(txnDataName(USER), range, null)), null, importMetadata); } @@ -446,15 +448,22 @@ public class TxnRead extends AbstractKeySorted implements Read // Metadata that is needed to be kept track of for ImportTxns public static class ImportMetadata { + private final UUID importID; private final TimeUUID planId; private final Long streamingEpoch; - public ImportMetadata(TimeUUID planId, Long streamingEpoch) + public ImportMetadata(UUID importID, TimeUUID planId, Long streamingEpoch) { + this.importID = importID; this.planId = planId; this.streamingEpoch = streamingEpoch; } + public UUID getImportID() + { + return importID; + } + public TimeUUID getPlanId() { return planId; @@ -473,7 +482,8 @@ public class TxnRead extends AbstractKeySorted implements Read public boolean equals(ImportMetadata that) { - return Objects.equals(this.planId, that.planId) + return Objects.equals(this.importID, that.importID) + && Objects.equals(this.planId, that.planId) && (Objects.equals(this.streamingEpoch, that.streamingEpoch)); } @@ -488,12 +498,14 @@ public class TxnRead extends AbstractKeySorted implements Read @Override public void serialize(ImportMetadata importMetadata, DataOutputPlus out, Version version) throws IOException { + UUIDSerializer.serializer.serialize(importMetadata.importID, out); TimeUUID.Serializer.instance.serialize(importMetadata.planId, out); out.writeLong(importMetadata.streamingEpoch); } public void skip(DataInputPlus in, Version version) throws IOException { + UUIDSerializer.serializer.skip(in); TimeUUID.Serializer.instance.skip(in); in.readLong(); } @@ -501,15 +513,17 @@ public class TxnRead extends AbstractKeySorted implements Read @Override public ImportMetadata deserialize(DataInputPlus in, Version version) throws IOException { + UUID importID = UUIDSerializer.serializer.deserialize(in); TimeUUID planId = TimeUUID.Serializer.instance.deserialize(in); Long streamingEpoch = in.readLong(); - return new ImportMetadata(planId, streamingEpoch); + return new ImportMetadata(importID, planId, streamingEpoch); } @Override public long serializedSize(ImportMetadata importMetadata, Version version) { long size = 0; + size += UUIDSerializer.serializer.serializedSize(importMetadata.importID); size += TimeUUID.Serializer.instance.serializedSize(importMetadata.planId); size += TypeSizes.LONG_SIZE; return size; 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 c11915d541..e232060a19 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java @@ -140,10 +140,6 @@ public class AccordImportSSTableTest extends TestBaseImpl } } - /** - * This might have a potential issue with us throwing an exception instead of having a custom class to notify - * that the read was a failure. - */ @Test public void testSSTableImportWithConcurrentTopologyChangeFails() throws Throwable { @@ -177,9 +173,14 @@ public class AccordImportSSTableTest extends TestBaseImpl Set paths = Set.of(file); Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true)) .isInstanceOf(RuntimeException.class) - .hasMessageContaining("Failed adding SSTables"); + .hasMessageContaining("Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator") + .cause() + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("SSTable import failed because of a concurrent topology change; please retry the operation"); + }); }, "importer"); + importer.start(); cluster.get(1).runOnInstance(() -> { @@ -228,8 +229,7 @@ public class AccordImportSSTableTest extends TestBaseImpl ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); Set paths = Set.of(file); Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true)) - .isInstanceOf(RuntimeException.class) - .cause(); + .isInstanceOf(RuntimeException.class); }); Iterable up = cluster.stream() @@ -380,7 +380,13 @@ public class AccordImportSSTableTest extends TestBaseImpl cluster.get(1).runOnInstance(() -> { ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); Set paths = Set.of(file); - Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true)); + + Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true)) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator") + .cause() + .isInstanceOf(RuntimeException.class) + .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 @@ -452,7 +458,13 @@ public class AccordImportSSTableTest extends TestBaseImpl cluster.get(1).runOnInstance(() -> { ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); Set paths = Set.of(file); - Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true)); + Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true)) + .isInstanceOf(RuntimeException.class) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator") + .cause() + .isInstanceOf(RuntimeException.class) + .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