Merge e28556f71c into 10557d7ffe

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/apache/cassandra-accord.git
+	url = https://github.com/alanwang67/cassandra-accord.git
-	branch = trunk
+	branch = CASSANDRA-20595

modules/accord

@@ -1 +1 @@
-Subproject commit fff32de2e915772fbc70d16b1c32346313877838
+Subproject commit e2352bc34edaf9e8662068ba163519c7ba2a50fb

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -948,7 +948,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
                                                                         .build());
     }
 
-    Descriptor getUniqueDescriptorFor(Descriptor descriptor, File targetDirectory)
+    public Descriptor getUniqueDescriptorFor(Descriptor descriptor, File targetDirectory)
     {
         Descriptor newDescriptor;
         do
@@ -2208,6 +2208,31 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
             CacheService.instance.invalidateCounterCacheForCf(metadata());
     }
 
+    public void invalidateRowAndCounterCache(Collection<SSTableReader> sstables, Consumer<Integer> onRowCacheInvalidation, Consumer<Integer> onCounterCacheInvalidation)
+    {
+        if (isRowCacheEnabled() || metadata().isCounter())
+        {
+            List<Bounds<Token>> boundsToInvalidate = new ArrayList<>(sstables.size());
+            sstables.forEach(sstable -> boundsToInvalidate.add(new Bounds<>(sstable.getFirst().getToken(), sstable.getLast().getToken())));
+            Set<Bounds<Token>> nonOverlappingBounds = Bounds.getNonOverlappingBounds(boundsToInvalidate);
+
+            if (isRowCacheEnabled())
+            {
+                int invalidatedKeys = invalidateRowCache(nonOverlappingBounds);
+                if (invalidatedKeys > 0)
+                    onRowCacheInvalidation.accept(invalidatedKeys);
+            }
+
+            if (metadata().isCounter())
+            {
+                int invalidatedKeys = invalidateCounterCache(nonOverlappingBounds);
+                if (invalidatedKeys > 0)
+                    onCounterCacheInvalidation.accept(invalidatedKeys);
+            }
+        }
+    }
+
+
     public int invalidateRowCache(Collection<Bounds<Token>> boundsToInvalidate)
     {
         int invalidatedKeys = 0;

src/java/org/apache/cassandra/db/Directories.java

@@ -74,6 +74,7 @@ import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.snapshot.SnapshotManifest;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.TimeUUID;
 
 import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
 
@@ -117,6 +118,7 @@ public class Directories
     public static final String BACKUPS_SUBDIR = "backups";
     public static final String SNAPSHOT_SUBDIR = "snapshots";
     public static final String TMP_SUBDIR = "tmp";
+    public static final String PENDING_SUBDIR = "pending";
     public static final String SECONDARY_INDEX_NAME_SEPARATOR = ".";
     public static final String TABLE_DIRECTORY_NAME_SEPARATOR = "-";
 
@@ -316,10 +318,7 @@ public class Directories
         if (dataDirectory != null)
             for (File dir : dataPaths)
             {
-                // Note that we must compare absolute paths (not canonical) here since keyspace directories might be symlinks
+                if (dataDirectory.contains(dir))
-                Path dirPath = dir.toAbsolute().toPath();
-                Path locationPath = dataDirectory.location.toAbsolute().toPath();
-                if (dirPath.startsWith(locationPath))
                     return dir;
             }
         return null;
@@ -726,6 +725,33 @@ public class Directories
         return new File(snapshotDir, "schema.cql");
     }
 
+    @VisibleForTesting
+    public Set<File> getPendingLocations()
+    {
+        Set<File> result = new HashSet<>();
+        for (DataDirectory dataDirectory : dataDirectories.getAllDirectories())
+        {
+            for (File dir : dataPaths)
+            {
+                if (!dataDirectory.contains(dir))
+                    continue;
+                result.add(getOrCreate(dir, PENDING_SUBDIR));
+            }
+        }
+        return result;
+    }
+
+    public File getPendingLocationForDisk(DataDirectory dataDirectory, TimeUUID planId)
+    {
+        for (File dir : dataPaths)
+        {
+            if (!dataDirectory.contains(dir))
+                continue;
+            return getOrCreate(dir, PENDING_SUBDIR, planId.toString());
+        }
+        throw new RuntimeException("Could not find pending location");
+    }
+
     public static File getBackupsDirectory(Descriptor desc)
     {
         return getBackupsDirectory(desc.directory);
@@ -814,6 +840,13 @@ public class Directories
             this.location = new File(location);
         }
 
+        public boolean contains(File file)
+        {
+            // Note that we must compare absolute paths (not canonical) here since keyspace directories might be symlinks
+            Path path = file.toAbsolute().toPath();
+            return path.startsWith(location.toAbsolute().toPath());
+        }
+
         public long getAvailableSpace()
         {
             long availableSpace = PathUtils.tryGetSpace(location.toPath(), FileStore::getUsableSpace) - DatabaseDescriptor.getMinFreeSpacePerDriveInBytes();

src/java/org/apache/cassandra/db/SSTableImporter.java

@@ -45,8 +45,10 @@ import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.File;
+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.utils.OutputHandler;
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.concurrent.Refs;
@@ -80,6 +82,8 @@ public class SSTableImporter
         UUID importID = UUID.randomUUID();
         logger.info("[{}] Loading new SSTables for {}/{}: {}", importID, cfs.getKeyspaceName(), cfs.getTableName(), options);
 
+        TableMetadata metadata = cfs.metadata();
+        boolean isAccordEnabled = metadata.isAccordEnabled();
         List<Pair<Directories.SSTableLister, String>> listers = getSSTableListers(options.srcPaths);
 
         Set<Descriptor> currentDescriptors = new HashSet<>();
@@ -178,7 +182,10 @@ public class SSTableImporter
                     Descriptor newDescriptor = cfs.getUniqueDescriptorFor(entry.getKey(), targetDir);
                     maybeMutateMetadata(entry.getKey(), options);
                     movedSSTables.add(new MovedSSTable(newDescriptor, entry.getKey(), entry.getValue()));
-                    SSTableReader sstable = SSTableReader.moveAndOpenSSTable(cfs, entry.getKey(), newDescriptor, entry.getValue(), options.copyData);
+                    // Don't move tracked SSTables, since that will move them to the live set on bounce
+                    SSTableReader sstable = isAccordEnabled
+                                            ? SSTableReader.open(cfs, oldDescriptor, metadata.ref)
+                                            : SSTableReader.moveAndOpenSSTable(cfs, oldDescriptor, newDescriptor, entry.getValue(), options.copyData);
                     newSSTablesPerDirectory.add(sstable);
                 }
                 catch (Throwable t)
@@ -228,7 +235,10 @@ public class SSTableImporter
             if (!cfs.indexManager.validateSSTableAttachedIndexes(newSSTables, false, options.validateIndexChecksum))
                 cfs.indexManager.buildSSTableAttachedIndexesBlocking(newSSTables);
 
-            cfs.getTracker().addSSTables(newSSTables);
+            if (isAccordEnabled)
+                AccordService.instance().executeTransfer(importID, cfs.keyspace.getName(), newSSTables, metadata);
+            else
+                cfs.getTracker().addSSTables(newSSTables);
             for (SSTableReader reader : newSSTables)
             {
                 if (options.invalidateCaches && cfs.isRowCacheEnabled())
@@ -237,8 +247,16 @@ public class SSTableImporter
         }
         catch (Throwable t)
         {
-            logger.error("[{}] Failed adding SSTables", importID, t);
+            if (isAccordEnabled)
-            throw new RuntimeException("Failed adding SSTables", t);
+            {
+                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());

src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java

@@ -45,6 +45,7 @@ import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.SequentialWriterOption;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.streaming.ProgressInfo;
+import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamReceiver;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.streaming.messages.StreamMessageHeader;
@@ -159,10 +160,14 @@ public class CassandraEntireSSTableStreamReader implements IStreamReader
 
     private File getDataDir(ColumnFamilyStore cfs, long totalSize) throws IOException
     {
+        boolean performingAccordBulkDataImport = cfs.metadata().isAccordEnabled() && session.streamOperation() == StreamOperation.ACCORD_SSTABLE_IMPORT;
         Directories.DataDirectory localDir = cfs.getDirectories().getWriteableLocation(totalSize);
         if (localDir == null)
             throw new IOException(format("Insufficient disk space to store %s", prettyPrintMemory(totalSize)));
 
+        if (performingAccordBulkDataImport)
+            return cfs.getDirectories().getPendingLocationForDisk(localDir, session.planId());
+
         File dir = cfs.getDirectories().getLocationForDisk(cfs.getDiskBoundaries().getCorrectDiskForKey(header.firstKey));
 
         if (dir == null)

src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java

@@ -47,6 +47,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterator;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.UnknownColumnException;
+import org.apache.cassandra.io.sstable.RangeAwarePendingSSTableWriter;
 import org.apache.cassandra.io.sstable.RangeAwareSSTableWriter;
 import org.apache.cassandra.io.sstable.SSTableMultiWriter;
 import org.apache.cassandra.io.sstable.SSTableSimpleIterator;
@@ -61,6 +62,7 @@ import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.streaming.ProgressInfo;
+import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamReceivedOutOfTokenRangeException;
 import org.apache.cassandra.streaming.StreamReceiver;
 import org.apache.cassandra.streaming.StreamSession;
@@ -185,7 +187,15 @@ public class CassandraStreamReader implements IStreamReader
         StreamReceiver streamReceiver = session.getAggregator(tableId);
         Preconditions.checkState(streamReceiver instanceof CassandraStreamReceiver);
         ILifecycleTransaction txn = createTxn();
-        RangeAwareSSTableWriter writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
+        RangeAwareSSTableWriter writer;
+        if (session.streamOperation() == StreamOperation.ACCORD_SSTABLE_IMPORT)
+        {
+            Preconditions.checkState(cfs.metadata().isAccordEnabled());
+            writer = new RangeAwarePendingSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()), session.planId());
+        }
+        else
+            writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, false, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
+
         return new SSTableTxnSingleStreamWriter(txn, writer);
     }
 

src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java

@@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory;
 
 import accord.primitives.Ranges;
 import accord.primitives.TxnId;
+import accord.utils.Invariants;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ColumnFamilyStore;
@@ -51,9 +52,11 @@ import org.apache.cassandra.io.sstable.SSTableTxnSingleStreamWriter;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.service.accord.IAccordService;
+import org.apache.cassandra.service.accord.PendingLocalTransfer;
 import org.apache.cassandra.service.accord.TimeOnlyRequestBookkeeping.LatencyRequestBookkeeping;
 import org.apache.cassandra.service.accord.topology.AccordTopology;
 import org.apache.cassandra.streaming.IncomingStream;
+import org.apache.cassandra.streaming.StreamOperation;
 import org.apache.cassandra.streaming.StreamReceiver;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.tcm.ClusterMetadata;
@@ -115,6 +118,7 @@ public class CassandraStreamReceiver implements StreamReceiver
         return (CassandraIncomingFile) stream;
     }
 
+    // This method is called for every SSTable within a stream
     @Override
     public synchronized void received(IncomingStream stream)
     {
@@ -242,6 +246,7 @@ public class CassandraStreamReceiver implements StreamReceiver
         {
             if (requiresWritePath)
             {
+                Invariants.require(session.streamOperation() != StreamOperation.ACCORD_SSTABLE_IMPORT);
                 sendThroughWritePath(cfs, readers);
             }
             else
@@ -257,6 +262,16 @@ public class CassandraStreamReceiver implements StreamReceiver
 
                 // add sstables (this will build non-SSTable-attached secondary indexes too, see CASSANDRA-10130)
                 logger.debug("[Stream #{}] Received {} sstables from {} ({})", session.planId(), readers.size(), session.peer, readers);
+
+                // Accord will coordinate marking these SSTables as live
+                if (session.streamOperation() == StreamOperation.ACCORD_SSTABLE_IMPORT)
+                {
+                    Preconditions.checkState(cfs.metadata().isAccordEnabled());
+                    PendingLocalTransfer transfer = new PendingLocalTransfer(cfs.metadata().id, session.planId(), sstables);
+                    AccordService.instance().receivedSSTableImport(transfer);
+                    return;
+                }
+
                 cfs.addSSTables(readers);
 
                 //invalidate row and counter cache

src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java

@@ -0,0 +1,44 @@
+/*
+ * 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.io.sstable;
+
+import java.io.IOException;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.SerializationHeader;
+import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.utils.TimeUUID;
+
+public class RangeAwarePendingSSTableWriter extends RangeAwareSSTableWriter
+{
+    private final TimeUUID planId;
+
+    public RangeAwarePendingSSTableWriter(ColumnFamilyStore cfs, long estimatedKeys, long repairedAt, TimeUUID pendingRepair, boolean isTransient, SSTableFormat<?, ?> format, int sstableLevel, long totalSize, ILifecycleTransaction txn, SerializationHeader header, TimeUUID planId) throws IOException
+    {
+        super(cfs, estimatedKeys, repairedAt, pendingRepair, isTransient, format, sstableLevel, totalSize, txn, header);
+        this.planId = planId;
+    }
+
+    @Override
+    protected File currentFile()
+    {
+        return cfs.getDirectories().getPendingLocationForDisk(directories.get(currentIndex), planId);
+    }
+}

src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java

@@ -32,6 +32,7 @@ import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
 import org.apache.cassandra.io.sstable.format.SSTableFormat;
 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.utils.FBUtilities;
 import org.apache.cassandra.utils.TimeUUID;
@@ -39,7 +40,7 @@ import org.apache.cassandra.utils.TimeUUID;
 public class RangeAwareSSTableWriter implements SSTableMultiWriter
 {
     private final List<PartitionPosition> boundaries;
-    private final List<Directories.DataDirectory> directories;
+    protected final List<Directories.DataDirectory> directories;
     private final int sstableLevel;
     private final long estimatedKeys;
     private final long repairedAt;
@@ -48,7 +49,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
     private final SSTableFormat<?, ?> format;
     private final SerializationHeader header;
     private final ILifecycleTransaction txn;
-    private int currentIndex = -1;
+    protected int currentIndex = -1;
     public final ColumnFamilyStore cfs;
     private final List<SSTableMultiWriter> finishedWriters = new ArrayList<>();
     private SSTableMultiWriter currentWriter = null;
@@ -95,11 +96,16 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
             if (currentWriter != null)
                 finishedWriters.add(currentWriter);
 
-            Descriptor desc = cfs.newSSTableDescriptor(cfs.getDirectories().getLocationForDisk(directories.get(currentIndex)), format);
+            Descriptor desc = cfs.newSSTableDescriptor(currentFile(), format);
             currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, pendingRepair, isTransient, null, sstableLevel, header, txn);
         }
     }
 
+    protected File currentFile()
+    {
+        return cfs.getDirectories().getLocationForDisk(directories.get(currentIndex));
+    }
+
     public void append(UnfilteredRowIterator partition)
     {
         maybeSwitchWriter(partition.partitionKey());
@@ -139,7 +145,8 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
     public SSTableMultiWriter setOpenResult(boolean openResult)
     {
         finishedWriters.forEach((w) -> w.setOpenResult(openResult));
-        currentWriter.setOpenResult(openResult);
+        if (currentWriter != null)
+            currentWriter.setOpenResult(openResult);
         return this;
     }
 

src/java/org/apache/cassandra/net/Verb.java

@@ -84,6 +84,8 @@ import org.apache.cassandra.schema.SchemaVersionVerbHandler;
 import org.apache.cassandra.service.EchoVerbHandler;
 import org.apache.cassandra.service.SnapshotVerbHandler;
 import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.LocalTransfers;
+import org.apache.cassandra.service.accord.TransferFailed;
 import org.apache.cassandra.service.accord.debug.AccordRemoteTracing;
 import org.apache.cassandra.service.accord.interop.AccordInteropApply;
 import org.apache.cassandra.service.accord.interop.AccordInteropRead;
@@ -385,6 +387,9 @@ public enum Verb
     DICTIONARY_UPDATE_RSP           (171, P1, rpcTimeout,   REQUEST_RESPONSE,  () -> NoPayload.serializer,                  RESPONSE_HANDLER                             ),
     DICTIONARY_UPDATE_REQ           (172, P1, rpcTimeout, MISC, () -> CompressionDictionaryUpdateMessage.serializer, () -> CompressionDictionaryUpdateVerbHandler.instance, DICTIONARY_UPDATE_RSP ),
 
+    TRACKED_TRANSFER_FAILED_RSP     (914, P1, repairTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER),
+    TRACKED_TRANSFER_FAILED_REQ     (915, P1, repairTimeout, ANTI_ENTROPY, () -> TransferFailed.serializer, () -> LocalTransfers.verbHandler, TRACKED_TRANSFER_FAILED_RSP),
+
     // generic failure response
     FAILURE_RSP            (99,  P0, noTimeout,       REQUEST_RESPONSE,  () -> RequestFailure.serializer,            RESPONSE_HANDLER                             ),
 

src/java/org/apache/cassandra/service/accord/AccordService.java

@@ -25,6 +25,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -39,6 +40,7 @@ import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import com.google.common.primitives.Ints;
 
 import org.agrona.collections.Long2LongHashMap;
@@ -102,6 +104,7 @@ import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.db.SystemKeyspace.BootstrapState;
 import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.exceptions.RequestExecutionException;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.journal.Descriptor;
 import org.apache.cassandra.journal.Params;
 import org.apache.cassandra.locator.InetAddressAndPort;
@@ -191,6 +194,8 @@ import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.DRAIN;
 import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.COMPLETED;
 import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadBookkeeping;
 import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteBookkeeping;
+import static org.apache.cassandra.service.accord.CoordinatedTransfer.getNodeStreamingContext;
+import static org.apache.cassandra.service.accord.CoordinatedTransfer.getTokenRangeSpanningSSTables;
 import static org.apache.cassandra.service.accord.topology.AccordTopology.tcmIdToAccord;
 import static org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.getTableMetadata;
 import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@@ -1436,4 +1441,25 @@ public class AccordService implements IAccordService, Shutdownable
     {
         return journal.configuration();
     }
+
+    public void executeTransfer(UUID importID, String keyspace, Set<SSTableReader> sstables, TableMetadata metadata)
+    {
+        logger.info("Creating Accord bulk transfer for keyspace '{}' table '{}' SSTables {}...", keyspace, metadata.name, sstables);
+
+        Topology topology = topology().current();
+
+        Map<InetAddressAndPort, CoordinatedTransfer.NodeStreamingMetadata> nodeStreamingContext = getNodeStreamingContext(sstables, topology, endpointMapper);
+
+        Preconditions.checkArgument(!nodeStreamingContext.isEmpty());
+
+        CoordinatedTransfer transfer = new CoordinatedTransfer(importID, metadata, nodeStreamingContext, topology.epoch(), getTokenRangeSpanningSSTables(sstables, metadata));
+
+        transfer.execute();
+    }
+
+    public void receivedSSTableImport(PendingLocalTransfer transfer)
+    {
+        logger.info("Received SSTables in pending directory");
+        LocalTransfers.instance().received(transfer);
+    }
 }

src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java

@@ -0,0 +1,418 @@
+/*
+ * 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.service.accord;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.local.Node;
+import accord.local.Node.Id;
+import accord.primitives.Ranges;
+import accord.primitives.Txn;
+import accord.topology.Topology;
+import accord.utils.Invariants;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
+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;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.NoPayload;
+import org.apache.cassandra.net.RequestCallbackWithFailure;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.accord.api.TokenKey;
+import org.apache.cassandra.service.accord.serializers.TableMetadatas;
+import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys;
+import org.apache.cassandra.service.accord.topology.AccordEndpointMapper;
+import org.apache.cassandra.service.accord.txn.TxnQuery;
+import org.apache.cassandra.service.accord.txn.TxnRead;
+import org.apache.cassandra.service.accord.txn.TxnResult;
+import org.apache.cassandra.streaming.OutgoingStream;
+import org.apache.cassandra.streaming.StreamException;
+import org.apache.cassandra.streaming.StreamOperation;
+import org.apache.cassandra.streaming.StreamPlan;
+import org.apache.cassandra.streaming.StreamResultFuture;
+import org.apache.cassandra.streaming.StreamState;
+import org.apache.cassandra.transport.Dispatcher;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Throwables;
+import org.apache.cassandra.utils.TimeUUID;
+import org.apache.cassandra.utils.concurrent.AsyncFuture;
+
+import static accord.primitives.Txn.Kind.Read;
+
+public class CoordinatedTransfer
+{
+    private static final Logger logger = LoggerFactory.getLogger(CoordinatedTransfer.class);
+
+    String logPrefix()
+    {
+        return String.format("[CoordinatedTransfer #%s]", importID.toString());
+    }
+
+    private final UUID importID;
+    private final TableMetadata tableMetadata;
+    private final long streamingEpoch;
+    private final TokenRange allSSTableRanges;
+
+    final Map<InetAddressAndPort, NodeStreamingMetadata> nodeStreamingContext;
+    SingleTransferResult streamResult = SingleTransferResult.Init();
+
+    public CoordinatedTransfer(UUID importID, TableMetadata tableMetadata, Map<InetAddressAndPort, NodeStreamingMetadata> nodeStreamingContext, long streamingEpoch, TokenRange allSSTableRanges)
+    {
+        this.importID = importID;
+        this.tableMetadata = tableMetadata;
+        this.nodeStreamingContext = nodeStreamingContext;
+        this.streamingEpoch = streamingEpoch;
+        this.allSSTableRanges = allSSTableRanges;
+    }
+
+    public UUID importID()
+    {
+        return importID;
+    }
+
+    void execute()
+    {
+        logger.debug("{} Executing Accord bulk transfer {}", logPrefix(), this);
+        LocalTransfers.instance().save(this);
+        stream();
+
+        try
+        {
+            performImportTxn();
+
+        }
+        catch (Exception 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, 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<TxnResult> accordResult = AccordService.instance().coordinateAsync(tableMetadata.epoch.getEpoch(), txn, ConsistencyLevel.ALL, Dispatcher.RequestTime.forImmediateExecution());
+        accordResult.awaitAndGet();
+    }
+
+    private void stream()
+    {
+        Future<Void> stream = LocalTransfers.instance().executor.submit(() -> {
+            SingleTransferResult result;
+            try
+            {
+                result = streamTask();
+            }
+            catch (StreamException | ExecutionException | InterruptedException | TimeoutException e)
+            {
+                Throwable cause = e instanceof ExecutionException ? e.getCause() : e;
+                streamResult = SingleTransferResult.streamFailed(streamResult, streamResult.planId);
+                throw Throwables.unchecked(cause);
+            }
+
+            streamResult = result;
+            logger.info("{} Completed streaming to all nodes", logPrefix());
+            return null;
+        });
+
+        try
+        {
+            stream.get();
+            logger.info("{} All streaming completed successfully", logPrefix());
+        }
+        catch (InterruptedException | ExecutionException e)
+        {
+            logger.error("{} Failed transfer due to", logPrefix(), e);
+            Throwable cause = e instanceof ExecutionException ? e.getCause() : e;
+            maybeCleanupFailedStreams(cause);
+            throw new RuntimeException("Failed streaming", Throwables.unchecked(cause));
+        }
+    }
+
+    private SingleTransferResult streamTask() throws StreamException, ExecutionException, InterruptedException, TimeoutException
+    {
+        StreamPlan plan = new StreamPlan(StreamOperation.ACCORD_SSTABLE_IMPORT);
+
+        // No need to flush, only using non-live SSTables already on disk
+        plan.flushBeforeTransfer(false);
+
+        for (Map.Entry<InetAddressAndPort, NodeStreamingMetadata> entry : nodeStreamingContext.entrySet())
+        {
+            InetAddressAndPort to = entry.getKey();
+            NodeStreamingMetadata sstablesForNode = entry.getValue();
+            List<Range<Token>> ranges = sstablesForNode.ranges;
+
+            for (Map.Entry<SSTableReader, List<SSTableReader.PartitionPositionBounds>> positionsForSSTables : sstablesForNode.positionsForSSTables.entrySet())
+            {
+                SSTableReader sstable = positionsForSSTables.getKey();
+                List<SSTableReader.PartitionPositionBounds> positions = positionsForSSTables.getValue();
+                long estimatedKeys = sstable.estimatedKeysForRanges(ranges);
+                OutgoingStream stream = new CassandraOutgoingFile(StreamOperation.ACCORD_SSTABLE_IMPORT, sstable.ref(), positions, ranges, estimatedKeys);
+                plan.transferStreams(to, Collections.singleton(stream));
+            }
+        }
+
+        long timeout = DatabaseDescriptor.getStreamTransferTaskTimeout().toMilliseconds();
+
+        logger.info("{} Starting streaming transfer", logPrefix());
+        streamResult = SingleTransferResult.streaming(streamResult, plan.planId());
+        StreamResultFuture execute = plan.execute();
+        StreamState state;
+        try
+        {
+            state = execute.get(timeout, TimeUnit.MILLISECONDS);
+            logger.debug("{} Completed streaming transfer", logPrefix());
+        }
+        catch (InterruptedException | ExecutionException | TimeoutException e)
+        {
+            logger.error("Stream session failed with error", e);
+            throw e;
+        }
+
+        if (state.hasFailedSession() || state.hasAbortedSession())
+            throw new StreamException(state, "Stream failed due to failed or aborted sessions");
+
+        return SingleTransferResult.streamComplete(streamResult, plan.planId());
+    }
+
+    /**
+     * This shouldn't throw an exception, even if we fail to notify peers of the streaming failure.
+     */
+    private void maybeCleanupFailedStreams(Throwable cause)
+    {
+        try
+        {
+            notifyFailure();
+            LocalTransfers.instance().schedulePendingLocalTransferCleanup(streamResult.planId());
+            LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this);
+        }
+        catch (Throwable t)
+        {
+            if (cause != null)
+                t.addSuppressed(cause);
+            logger.error("{} Failed to notify peers of stream failure", logPrefix(), t);
+        }
+    }
+
+    private void notifyFailure() throws ExecutionException, InterruptedException
+    {
+        class NotifyFailure extends AsyncFuture<Void> implements RequestCallbackWithFailure<NoPayload>
+        {
+            final Set<InetAddressAndPort> responses = ConcurrentHashMap.newKeySet(nodeStreamingContext.size());
+
+            @Override
+            public void onResponse(Message<NoPayload> msg)
+            {
+                responses.remove(msg.from());
+                if (responses.isEmpty())
+                    trySuccess(null);
+            }
+
+            @Override
+            public void onFailure(InetAddressAndPort from, RequestFailure failure)
+            {
+                // Log but don't fail - best effort cleanup
+                logger.warn("{} Failed to notify {} of transfer failure: {}", logPrefix(), from, failure);
+                responses.remove(from);
+                if (responses.isEmpty())
+                    trySuccess(null);
+            }
+        }
+
+        Map<InetAddressAndPort, Message<TransferFailed>> msgs = new HashMap<>();
+        NotifyFailure notifyFailure = new NotifyFailure();
+
+        TimeUUID planId = streamResult.planId;
+        Invariants.require(planId != null);
+
+        for (InetAddressAndPort to : nodeStreamingContext.keySet())
+        {
+            // Coordinator cleans up CoordinatedTransfer and PendingLocalTransfer separately, does not need to notify
+            if (FBUtilities.getBroadcastAddressAndPort().equals(to))
+                continue;
+
+            logger.debug("{}, Notifying {} of transfer failure for plan {}", logPrefix(), to, streamResult.planId);
+            notifyFailure.responses.add(to);
+            msgs.put(to, Message.out(Verb.TRACKED_TRANSFER_FAILED_REQ, new TransferFailed(streamResult.planId)));
+        }
+
+        for (Map.Entry<InetAddressAndPort,Message<TransferFailed>> entry : msgs.entrySet())
+            MessagingService.instance().sendWithCallback(entry.getValue(), entry.getKey(), notifyFailure);
+
+        if (!msgs.isEmpty())
+            notifyFailure.get();
+    }
+
+    public static class NodeStreamingMetadata
+    {
+        final Node.Id id;
+        final List<Range<Token>> ranges;
+        final Map<SSTableReader, List<SSTableReader.PartitionPositionBounds>> positionsForSSTables;
+
+        public NodeStreamingMetadata(Node.Id id, Map<SSTableReader, List<SSTableReader.PartitionPositionBounds>> positionsForSSTables, List<Range<Token>> ranges)
+        {
+            this.id = id;
+            this.positionsForSSTables = positionsForSSTables;
+            this.ranges = ranges;
+        }
+    }
+
+    static class SingleTransferResult
+    {
+        enum State
+        {
+            INIT,
+            STREAMING,
+            STREAM_FAILED,
+            STREAM_COMPLETE;
+        }
+
+        final State state;
+        private final TimeUUID planId;
+
+        @VisibleForTesting
+        SingleTransferResult(State state, TimeUUID planId)
+        {
+            this.state = state;
+            this.planId = planId;
+        }
+
+        public static SingleTransferResult Init()
+        {
+            return new SingleTransferResult(State.INIT, null);
+        }
+
+        public static SingleTransferResult streaming(SingleTransferResult from, TimeUUID planId)
+        {
+            Invariants.require(from.state == State.INIT);
+            return new SingleTransferResult(State.STREAMING, planId);
+        }
+
+        public static SingleTransferResult streamComplete(SingleTransferResult from, TimeUUID planId)
+        {
+            Invariants.require(from.state == State.STREAMING);
+            return new SingleTransferResult(State.STREAM_COMPLETE, planId);
+        }
+
+        public static SingleTransferResult streamFailed(SingleTransferResult from, TimeUUID planId)
+        {
+            Invariants.require(from.state == State.STREAMING);
+            return new SingleTransferResult(State.STREAM_FAILED, planId);
+        }
+
+        public TimeUUID planId()
+        {
+            return planId;
+        }
+
+        @Override
+        public String toString()
+        {
+            return "SingleTransferResult{" +
+                   "state=" + state +
+                   ", planId=" + planId +
+                   '}';
+        }
+    }
+
+    public static TokenRange getTokenRangeSpanningSSTables(Collection<SSTableReader> sstables, TableMetadata metadata)
+    {
+        TokenKey minTokenKey = null;
+        TokenKey maxTokenKey = null;
+        for (SSTableReader sstable : sstables)
+        {
+            TokenKey startTokenKey = TokenKey.before(metadata.id, sstable.getFirst().getToken());
+            TokenKey endTokenKey = new TokenKey(metadata.id, sstable.getLast().getToken());
+            if (minTokenKey == null)
+                minTokenKey = startTokenKey;
+            else if (minTokenKey.compareTo(startTokenKey) > 0)
+                minTokenKey = startTokenKey;
+
+            if (maxTokenKey == null)
+                maxTokenKey = endTokenKey;
+            else if (maxTokenKey.compareTo(endTokenKey) < 0)
+                maxTokenKey = endTokenKey;
+        }
+
+        return new TokenRange(minTokenKey, maxTokenKey);
+    }
+
+    public static Map<InetAddressAndPort, NodeStreamingMetadata> getNodeStreamingContext(Collection<SSTableReader> sstables, Topology topology, AccordEndpointMapper endpointMapper)
+    {
+        Map<InetAddressAndPort, NodeStreamingMetadata> nodeStreamingContext = new HashMap<>();
+
+        for (Id nodeId : topology.nodes())
+        {
+            // Transform the ranges that each node owns to an input that can be used by
+            // getPositionsForRanges
+            Ranges rangesForNode = topology.rangesForNode(nodeId);
+            List<Range<Token>> ranges = new ArrayList<>();
+            for (accord.primitives.Range range : rangesForNode)
+                ranges.add(((TokenRange) range).toKeyspaceRange());
+
+            // Map from SSTables to the portion of the SSTable that the node owns
+            Map<SSTableReader, List<SSTableReader.PartitionPositionBounds>> positionsForSSTables = new HashMap<>();
+
+            for (SSTableReader sstable : sstables)
+            {
+                List<SSTableReader.PartitionPositionBounds> partition = sstable.getPositionsForRanges(ranges);
+                if (!partition.isEmpty())
+                    positionsForSSTables.put(sstable, partition);
+            }
+
+            if (!positionsForSSTables.isEmpty())
+            {
+                InetAddressAndPort endpoint = endpointMapper.mappedEndpointOrNull(nodeId);
+                if (endpoint == null)
+                    throw new RuntimeException("No endpoint for " + nodeId);
+                nodeStreamingContext.put(endpoint, new NodeStreamingMetadata(nodeId, positionsForSSTables, NormalizedRanges.normalizedRanges(ranges)));
+            }
+        }
+
+        return nodeStreamingContext;
+    }
+}

src/java/org/apache/cassandra/service/accord/IAccordService.java

@@ -21,6 +21,8 @@ package org.apache.cassandra.service.accord;
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.BiConsumer;
@@ -57,10 +59,12 @@ import accord.utils.async.AsyncResult;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.exceptions.RequestExecutionException;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.journal.Params;
 import org.apache.cassandra.net.IVerbHandler;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.accord.api.AccordScheduler;
 import org.apache.cassandra.service.accord.api.AccordTopologySorter;
 import org.apache.cassandra.service.accord.topology.AccordEndpointMapper;
@@ -193,6 +197,10 @@ public interface IAccordService
 
     Node node();
 
+    void executeTransfer(UUID importID, String keyspace, Set<SSTableReader> sstables, TableMetadata metadata);
+
+    void receivedSSTableImport(PendingLocalTransfer transfer);
+
     /**
      * Ensure Accord's hlc is at least larger than this for anything accepted at this node
      */
@@ -372,6 +380,18 @@ public interface IAccordService
             return null;
         }
 
+        @Override
+        public void executeTransfer(UUID importID, String keyspace, Set<SSTableReader> sstables, TableMetadata metadata)
+        {
+            throw new UnsupportedOperationException("Cannot import SSTables through Accord when accord.enabled = false in cassandra.yaml");
+        }
+
+        @Override
+        public void receivedSSTableImport(PendingLocalTransfer transfer)
+        {
+            throw new UnsupportedOperationException("Cannot import SSTables through Accord when accord.enabled = false in cassandra.yaml");
+        }
+
         @Override
         public void ensureMinHlc(long minHlc)
         {
@@ -564,6 +584,18 @@ public interface IAccordService
             return delegate.node();
         }
 
+        @Override
+        public void executeTransfer(UUID importID, String keyspace, Set<SSTableReader> sstables, TableMetadata metadata)
+        {
+            delegate.executeTransfer(importID, keyspace, sstables, metadata);
+        }
+
+        @Override
+        public void receivedSSTableImport(PendingLocalTransfer transfer)
+        {
+            delegate.receivedSSTableImport(transfer);
+        }
+
         @Override
         public void ensureMinHlc(long minHlc)
         {

src/java/org/apache/cassandra/service/accord/LocalTransfers.java

@@ -0,0 +1,205 @@
+/*
+ * 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.service.accord;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.base.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.ExecutorPlus;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.net.IVerbHandler;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.NoPayload;
+import org.apache.cassandra.service.accord.txn.TxnRead;
+import org.apache.cassandra.utils.ExecutorUtils;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
+
+public class LocalTransfers
+{
+    private static final Logger logger = LoggerFactory.getLogger(LocalTransfers.class);
+
+    private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+    // SSTable imports that we are coordinating
+    public final Map<UUID, CoordinatedTransfer> coordinating = new ConcurrentHashMap<>();
+
+    // Added when we have a streamed SSTable in our pending directory
+    public final Map<TimeUUID, PendingLocalTransfer> local = new ConcurrentHashMap<>();
+
+    final ExecutorPlus executor = executorFactory().pooled("LocalTrackedTransfers", Integer.MAX_VALUE);
+
+    public static final LocalTransfers instance = new LocalTransfers();
+    static LocalTransfers instance()
+    {
+        return instance;
+    }
+
+    void save(CoordinatedTransfer transfer)
+    {
+        lock.writeLock().lock();
+        try
+        {
+            CoordinatedTransfer existing = coordinating.put(transfer.importID(), transfer);
+            Preconditions.checkState(existing == null);
+        }
+        finally
+        {
+            lock.writeLock().unlock();
+        }
+    }
+
+    void received(PendingLocalTransfer transfer)
+    {
+        lock.writeLock().lock();
+        try
+        {
+            logger.debug("received: {}", transfer);
+            Preconditions.checkState(!transfer.sstables.isEmpty());
+
+            PendingLocalTransfer existing = local.put(transfer.planId, transfer);
+            Preconditions.checkState(existing == null);
+        }
+        finally
+        {
+            lock.writeLock().unlock();
+        }
+    }
+
+    private void purge(TimeUUID timeUUID)
+    {
+        lock.writeLock().lock();
+        try
+        {
+            PendingLocalTransfer transfer = local.get(timeUUID);
+            if (transfer == null)
+            {
+                logger.warn("Cannot purge unknown local pending transfer {}", transfer);
+                return;
+            }
+
+            logger.info("Cleaning up pending transfer {}", transfer);
+            // Delete the entire pending transfer directory /pending/<planId>/
+            if (!transfer.sstables.isEmpty())
+            {
+                Set<File> pendingDirs = new HashSet<>();
+                transfer.sstables.forEach(sstable -> {
+                    pendingDirs.add(sstable.descriptor.directory);
+                });
+
+                for (File pendingDir : pendingDirs)
+                {
+                    if (pendingDir.exists())
+                    {
+                        Preconditions.checkState(pendingDir.absolutePath().contains(transfer.planId.toString()));
+                        logger.debug("Deleting pending transfer directory: {}", pendingDir);
+                        pendingDir.deleteRecursive();
+                    }
+                }
+
+                local.remove(transfer.planId);
+            }
+        }
+        finally
+        {
+            lock.writeLock().unlock();
+        }
+    }
+
+    private void purge(CoordinatedTransfer transfer)
+    {
+        logger.info("Cleaning up completed coordinated transfer: {}", transfer);
+
+        lock.writeLock().lock();
+        try
+        {
+            coordinating.remove(transfer.importID());
+        }
+        finally
+        {
+            lock.writeLock().unlock();
+        }
+    }
+
+    void scheduleCoordinatedTransferCleanup(CoordinatedTransfer transfer)
+    {
+        executor.submit(() -> {
+            try
+            {
+                purge(transfer);
+            }
+            catch (Throwable t)
+            {
+                logger.error("Cleanup failed", t);
+            }
+        });
+    }
+
+    void schedulePendingLocalTransferCleanup(TimeUUID timeUUID)
+    {
+        executor.submit(() -> {
+            try
+            {
+                purge(timeUUID);
+            }
+            catch (Throwable t)
+            {
+                logger.error("Cleanup failed", t);
+            }
+        });
+    }
+
+    public void activatePendingTransfers(TxnRead.ImportMetadata metadata, long executeAtEpoch)
+    {
+        lock.readLock().lock();
+        try
+        {
+            PendingLocalTransfer pendingLocalTransfer = local.get(metadata.getPlanId());
+            if (pendingLocalTransfer != null)
+                pendingLocalTransfer.activate(metadata, executeAtEpoch);
+        }
+        finally
+        {
+            lock.readLock().unlock();
+        }
+    }
+
+    public void shutdownNowAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
+    {
+        ExecutorUtils.shutdownNowAndWait(timeout, unit, executor);
+    }
+
+    public static IVerbHandler<TransferFailed> verbHandler = message -> {
+        LocalTransfers.instance().purge(message.payload.planId);
+        MessagingService.instance().respond(NoPayload.noPayload, message);
+    };
+}

src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java

@@ -0,0 +1,163 @@
+/*
+ * 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.service.accord;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.function.Consumer;
+
+import javax.annotation.Nullable;
+
+import com.google.common.base.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.utils.Invariants;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+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;
+
+public class PendingLocalTransfer
+{
+    private static final Logger logger = LoggerFactory.getLogger(PendingLocalTransfer.class);
+
+    private String logPrefix()
+    {
+        return String.format("[PendingLocalTransfer #%s]", planId);
+    }
+
+    final TimeUUID planId;
+    final TableId tableId;
+    final Collection<SSTableReader> sstables;
+    final long createdAt = currentTimeMillis();
+    transient String keyspace;
+    @Nullable CountDownLatch latch;
+
+    volatile boolean activated = false;
+
+    public PendingLocalTransfer(TableId tableId, TimeUUID planId, Collection<SSTableReader> sstables)
+    {
+        Preconditions.checkState(!sstables.isEmpty());
+        this.tableId = tableId;
+        this.planId = planId;
+        this.sstables = sstables;
+        this.keyspace = Objects.requireNonNull(ColumnFamilyStore.getIfExists(tableId)).keyspace.getName();
+    }
+
+    // We maintain the invariant that only the coordinator registers the latch
+    public synchronized void registerLatch(CountDownLatch latch)
+    {
+        this.latch = latch;
+    }
+
+    /**
+     * Safely moves SSTables into the live set.
+     */
+    public synchronized void activate(TxnRead.ImportMetadata metadata, long executeAtEpoch)
+    {
+        if (activated)
+            return;
+
+        Invariants.require(metadata.getStreamingEpoch() == executeAtEpoch);
+        long startedActivation = currentTimeMillis();
+        logger.info("{} Activating transfer {}, {} ms since pending", logPrefix(), this, startedActivation - createdAt);
+        ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(tableId);
+        Preconditions.checkNotNull(cfs);
+        Preconditions.checkState(!sstables.isEmpty());
+
+        File dst = cfs.getDirectories().getDirectoryForNewSSTables();
+
+        dst.createFileIfNotExists();
+        Collection<SSTableReader> moved = new ArrayList<>(sstables.size());
+        Collection<Descriptor> movedDescriptors = new ArrayList<>(sstables.size());
+        for (SSTableReader sstable : sstables)
+        {
+            try
+            {
+                Descriptor newDescriptor = cfs.getUniqueDescriptorFor(sstable.descriptor, dst);
+                movedDescriptors.add(newDescriptor);
+                SSTableReader movedSSTable = SSTableReader.moveAndOpenSSTable(cfs, sstable.descriptor, newDescriptor, sstable.getComponents(), true);
+                moved.add(movedSSTable);
+            }
+            catch (Throwable t)
+            {
+                moved.forEach(s -> s.selfRef().release());
+                for (Descriptor descriptor : movedDescriptors)
+                    descriptor.getFormat().delete(descriptor);
+                throw new RuntimeException("Failed importing SSTables", t);
+            }
+        }
+
+        // Add all SSTables atomically
+        cfs.getTracker().addSSTables(moved);
+        activated = true;
+
+        Consumer<Integer> onRowCacheInvalidation = invalidatedKeys -> {
+            logger.debug("{} Invalidated {} row cache entries on table {}.{} after activating transfer",
+                         logPrefix(), invalidatedKeys, cfs.getKeyspaceName(), cfs.getTableName());
+        };
+        Consumer<Integer> onCounterCacheInvalidation = invalidatedKeys -> {
+            logger.debug("{} Invalidated {} counter cache entries on table {}.{} after activating transfer",
+                         logPrefix(), invalidatedKeys, cfs.getKeyspaceName(), cfs.getTableName());
+        };
+        cfs.invalidateRowAndCounterCache(moved, onRowCacheInvalidation, onCounterCacheInvalidation);
+
+        long finishedActivation = currentTimeMillis();
+        logger.info("{} Finished activating transfer {} in {} ms", logPrefix(), this, finishedActivation - startedActivation);
+
+        LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId);
+    }
+
+    @Override
+    public String toString()
+    {
+        return "PendingLocalTransfer{" +
+               "activated=" + activated +
+               ", keyspace='" + keyspace + '\'' +
+               ", createdAt=" + createdAt +
+               ", sstables=" + sstables +
+               ", tableId=" + tableId +
+               ", planId=" + planId +
+               '}';
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (o == null || getClass() != o.getClass()) return false;
+        PendingLocalTransfer transfer = (PendingLocalTransfer) o;
+        return Objects.equals(planId, transfer.planId) && Objects.equals(tableId, transfer.tableId) && Objects.equals(sstables, transfer.sstables);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(planId, tableId, sstables);
+    }
+}

src/java/org/apache/cassandra/service/accord/TransferFailed.java

@@ -0,0 +1,72 @@
+/*
+ * 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.service.accord;
+
+import java.io.IOException;
+
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.utils.TimeUUID;
+
+/**
+ * Notification from coordinator to replicas when a bulk data transfer fails, triggering cleanup of the pending
+ * transfer state.
+ * @see CoordinatedTransfer
+ * @see PendingLocalTransfer
+ */
+public class TransferFailed
+{
+    final TimeUUID planId;
+
+    public TransferFailed(TimeUUID planId)
+    {
+        this.planId = planId;
+    }
+
+    public static final IVersionedSerializer<TransferFailed> serializer = new IVersionedSerializer<TransferFailed>()
+    {
+        @Override
+        public void serialize(TransferFailed t, DataOutputPlus out, int version) throws IOException
+        {
+            TimeUUID.Serializer.instance.serialize(t.planId, out, version);
+        }
+
+        @Override
+        public TransferFailed deserialize(DataInputPlus in, int version) throws IOException
+        {
+            TimeUUID planId = TimeUUID.Serializer.instance.deserialize(in, version);
+            return new TransferFailed(planId);
+        }
+
+        @Override
+        public long serializedSize(TransferFailed t, int version)
+        {
+            return TimeUUID.Serializer.instance.serializedSize(t.planId, version);
+        }
+    };
+
+    @Override
+    public String toString()
+    {
+        return "TransferFailed{" +
+               "planId=" + planId +
+               '}';
+    }
+}

src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java

@@ -59,6 +59,7 @@ import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.service.accord.AccordExecutor;
+import org.apache.cassandra.service.accord.LocalTransfers;
 import org.apache.cassandra.service.accord.TokenRange;
 import org.apache.cassandra.service.accord.api.PartitionKey;
 import org.apache.cassandra.service.accord.api.TokenKey;
@@ -109,6 +110,13 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized<ReadC
         this.key = key;
     }
 
+    TxnNamedRead(int name, TokenRange range, ByteBuffer bytes)
+    {
+        super(bytes);
+        this.name = name;
+        this.key = range;
+    }
+
     public static TokenRange boundsAsAccordRange(AbstractBounds<PartitionPosition> range, TableId tableId)
     {
         // Should already have been unwrapped
@@ -402,6 +410,15 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized<ReadC
         return submit(executor, callable, callable);
     }
 
+    public AsyncChain<Data> performSSTableImport(AccordExecutor executor, TxnRead.ImportMetadata importMetadata, Timestamp executeAt)
+    {
+        Callable<Data> callable = () -> {
+            LocalTransfers.instance.activatePendingTransfers(importMetadata, executeAt.epoch());
+            return new TxnData();
+        };
+        return submit(executor, callable, callable);
+    }
+
     private AsyncChain<Data> submit(AccordExecutor executor, Callable<Data> readCallable, Object describe)
     {
         return executor.buildDebuggable(readCallable, describe);

src/java/org/apache/cassandra/service/accord/txn/TxnRead.java

@@ -22,6 +22,8 @@ import java.io.IOException;
 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;
@@ -50,17 +52,22 @@ import org.apache.cassandra.db.PartitionPosition;
 import org.apache.cassandra.db.PartitionRangeReadCommand;
 import org.apache.cassandra.db.ReadCommand;
 import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.io.ParameterisedVersionedSerializer;
+import org.apache.cassandra.io.VersionedSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.service.accord.AccordCommandStore;
 import org.apache.cassandra.service.accord.AccordExecutor;
+import org.apache.cassandra.service.accord.TokenRange;
 import org.apache.cassandra.service.accord.api.PartitionKey;
 import org.apache.cassandra.service.accord.serializers.TableMetadatas;
 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;
@@ -80,13 +87,14 @@ import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSi
 
 public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
 {
-    private static final TxnRead EMPTY_KEY = new TxnRead(TableMetadatas.none(), Domain.Key);
+    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);
+    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<TxnNamedRead> TXN_NAMED_READ_KEY_COMPARATOR = Comparator.comparing(a -> ((PartitionKey) a.key()));
     private static final byte TYPE_EMPTY_KEY = 0;
     private static final byte TYPE_EMPTY_RANGE = 1;
     private static final byte TYPE_NOT_EMPTY = 2;
+    private static final byte TYPE_IMPORT = 3;
 
     public static TxnRead empty(Domain domain)
     {
@@ -109,34 +117,40 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
     // Specifies the domain in case the TxnRead is empty and it can't be inferred
     private final Domain domain;
 
-    private TxnRead(TableMetadatas tables, Domain domain)
+    @Nullable
+    private final ImportMetadata importMetadata;
+
+    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 = importMetadata;
     }
 
-    private TxnRead(TableMetadatas tables, @Nonnull TxnNamedRead[] items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnRead(TableMetadatas tables, @Nonnull TxnNamedRead[] items, @Nullable ConsistencyLevel cassandraConsistencyLevel, @Nullable ImportMetadata importMetadata)
     {
         super(items, items[0].key().domain());
         this.tables = tables;
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read: %s", cassandraConsistencyLevel);
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
         this.domain = items[0].key().domain();
+        this.importMetadata = importMetadata;
         // TODO (expected): relax this condition, require only that it holds for each equal byte[]
         //  right now this means we don't permit two different range queries in the same transaction touching adjacent ranges
         //  this is a pretty weak restriction and doesn't interfere with current CQL capabilities, but should be addressed eventually
         require(domain == Domain.Key || ((Ranges)keys()).mergeTouching() == keys());
     }
 
-    private TxnRead(TableMetadatas tables, @Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnRead(TableMetadatas tables, @Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel cassandraConsistencyLevel, @Nullable ImportMetadata importMetadata)
     {
         super(items, items.get(0).key().domain());
         this.tables = tables;
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read: %s", cassandraConsistencyLevel);
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
         this.domain = items.get(0).key().domain();
+        this.importMetadata = importMetadata;
         require(domain == Domain.Key || ((Ranges)keys()).mergeTouching() == keys());
     }
 
@@ -151,7 +165,15 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
         if (items.isEmpty())
             return empty(domain);
         sortReads(items);
-        return new TxnRead(tables, items, consistencyLevel);
+        return new TxnRead(tables, items, consistencyLevel, null);
+    }
+
+    public static TxnRead createTxnRead(TableMetadatas tables, @Nonnull List<TxnNamedRead> 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<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel, TableMetadatasAndKeys.KeyCollector keyCollector)
@@ -163,13 +185,13 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
             reads.add(new TxnNamedRead(txnDataName(USER, i), keyCollector.collect(readCommand.metadata(), readCommand.partitionKey()), readCommand, keyCollector.tables));
         }
         sortReads(reads);
-        return new TxnRead(keyCollector.tables, reads, consistencyLevel);
+        return new TxnRead(keyCollector.tables, reads, consistencyLevel, null);
     }
 
     public static TxnRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel, TableMetadatasAndKeys tablesAndKeys)
     {
         TxnNamedRead read = new TxnNamedRead(txnDataName(CAS_READ), (PartitionKey) tablesAndKeys.keys.get(0), readCommand, tablesAndKeys.tables);
-        return new TxnRead(tablesAndKeys.tables, ImmutableList.of(read), consistencyLevel);
+        return new TxnRead(tablesAndKeys.tables, ImmutableList.of(read), consistencyLevel, null);
     }
 
     // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
@@ -179,12 +201,22 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
         List<TxnNamedRead> reads = new ArrayList<>(keys.size());
         for (int i = 0; i < keys.size(); i++)
             reads.add(new TxnNamedRead(txnDataName(USER, i), keys.get(i), null));
-        return new TxnRead(TableMetadatas.none(), reads, null);
+        return new TxnRead(TableMetadatas.none(), reads, null, null);
     }
 
     public static TxnRead createRangeRead(TableMetadatas tables, PartitionRangeReadCommand command, AbstractBounds<PartitionPosition> range, ConsistencyLevel consistencyLevel)
     {
-        return new TxnRead(tables, ImmutableList.of(new TxnNamedRead(txnDataName(USER), range, command, tables)), consistencyLevel);
+        return new TxnRead(tables, ImmutableList.of(new TxnNamedRead(txnDataName(USER), range, command, tables)), consistencyLevel, null);
+    }
+
+    // SSTable imports are performed through read range txn's. The value of the reads are not returned back to the client, rather
+    // 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, UUID importID, TimeUUID planId, long streamingEpoch)
+    {
+        ImportMetadata importMetadata = new ImportMetadata(importID, planId, streamingEpoch);
+        return new TxnRead(tables, ImmutableList.of(new TxnNamedRead(txnDataName(USER), range, null)), null, importMetadata);
     }
 
     public long estimatedSizeOnHeap()
@@ -291,7 +323,7 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
                 throw new UnhandledEnum(select.domain());
         }
 
-        return createTxnRead(tables, reads, cassandraConsistencyLevel, select.domain());
+        return createTxnRead(tables, reads, cassandraConsistencyLevel, select.domain(), importMetadata);
     }
 
     @Override
@@ -372,7 +404,8 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
                 break;
             }
         }
-        return createTxnRead(tables, reads, cassandraConsistencyLevel, that.domain);
+
+        return createTxnRead(tables, reads, cassandraConsistencyLevel, that.domain, importMetadata);
     }
 
     public void unmemoize()
@@ -394,6 +427,10 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
             return AsyncChains.success(new TxnData());
 
         AccordExecutor executor = ((AccordCommandStore)commandStore).executor();
+
+        if (isUsedForImport())
+            return items[0].performSSTableImport(executor, importMetadata, executeAt);
+
         if (items.length == 1 && key.equals(items[0].key()))
             return items[0].read(executor, tables, cassandraConsistencyLevel, key, executeAt);
 
@@ -406,12 +443,119 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
         return AsyncChains.reduce(results, Data::merge);
     }
 
+    // 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(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;
+        }
+
+        public Long getStreamingEpoch()
+        {
+            return streamingEpoch;
+        }
+
+        @Override
+        public boolean equals(Object that)
+        {
+            return that instanceof ImportMetadata && equals((ImportMetadata) that);
+        }
+
+        public boolean equals(ImportMetadata that)
+        {
+            return Objects.equals(this.importID, that.importID)
+                   && Objects.equals(this.planId, that.planId)
+                   && (Objects.equals(this.streamingEpoch, that.streamingEpoch));
+        }
+
+        @Override
+        public int hashCode()
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        public static final VersionedSerializer<ImportMetadata, Version> serializer = new VersionedSerializer<>()
+        {
+            @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();
+            }
+
+            @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(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;
+            }
+        };
+    }
+
+    @Override
+    public boolean isUsedForImport()
+    {
+        return importMetadata != null;
+    }
+
+    @Override
+    public long getImportStreamingEpoch()
+    {
+        if (importMetadata != null)
+            return importMetadata.streamingEpoch;
+        return -1L;
+    }
+
     public static final ParameterisedVersionedSerializer<TxnRead, TableMetadatasAndKeys, Version> serializer = new ParameterisedVersionedSerializer<>()
     {
         @Override
         public void serialize(TxnRead read, TableMetadatasAndKeys tablesAndKeys, DataOutputPlus out, Version version) throws IOException
         {
-            if (read.items.length > 0)
+            if (read.isUsedForImport())
+            {
+                out.write(TYPE_IMPORT);
+                serializeArray(read.items, tablesAndKeys, out, version, TxnNamedRead.serializer);
+                serializeNullable(read.cassandraConsistencyLevel, out, consistencyLevelSerializer);
+                serializeNullable(read.importMetadata, out, version, ImportMetadata.serializer);
+            }
+            else if (read.items.length > 0)
             {
                 out.write(TYPE_NOT_EMPTY);
                 serializeArray(read.items, tablesAndKeys, out, version, TxnNamedRead.serializer);
@@ -428,14 +572,20 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
             byte type = in.readByte();
             switch (type)
             {
-                default:
-                    throw new IllegalStateException("Unhandled type " + type);
                 case TYPE_EMPTY_KEY:
                 case TYPE_EMPTY_RANGE:
                     return;
                 case TYPE_NOT_EMPTY:
                     skipArray(tablesAndKeys, in, version, TxnNamedRead.serializer);
                     deserializeNullable(in, consistencyLevelSerializer);
+                    return;
+                case TYPE_IMPORT:
+                    skipArray(tablesAndKeys, in, version, TxnNamedRead.serializer);
+                    deserializeNullable(in, consistencyLevelSerializer);
+                    deserializeNullable(in, version, ImportMetadata.serializer);
+                    return;
+                default:
+                    throw new IllegalStateException("Unhandled type " + type);
             }
         }
 
@@ -453,9 +603,21 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
                 case TYPE_EMPTY_RANGE:
                     return EMPTY_RANGE;
                 case TYPE_NOT_EMPTY:
+                {
                     TxnNamedRead[] items = deserializeArray(tablesAndKeys, in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);
                     ConsistencyLevel consistencyLevel = deserializeNullable(in, consistencyLevelSerializer);
-                    return new TxnRead(tablesAndKeys.tables, items, consistencyLevel);
+                    return new TxnRead(tablesAndKeys.tables, items, consistencyLevel, null);
+                }
+                case TYPE_IMPORT:
+                {
+                    TxnNamedRead[] items = deserializeArray(tablesAndKeys, in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);
+                    ConsistencyLevel consistencyLevel = deserializeNullable(in, consistencyLevelSerializer);
+                    ImportMetadata importMetadata = deserializeNullable(in, version, ImportMetadata.serializer);
+                    if (items.length == 0)
+                        return new TxnRead(TableMetadatas.none(), Domain.Range, importMetadata);
+                    else
+                        return new TxnRead(tablesAndKeys.tables, items, consistencyLevel, importMetadata);
+                }
             }
         }
 
@@ -463,7 +625,13 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
         public long serializedSize(TxnRead read, TableMetadatasAndKeys tablesAndKeys, Version version)
         {
             long size = 1; // type
-            if (read.items.length > 0)
+            if (read.isUsedForImport())
+            {
+                size += serializedArraySize(read.items, tablesAndKeys, version, TxnNamedRead.serializer);
+                size += serializedNullableSize(read.cassandraConsistencyLevel, consistencyLevelSerializer);
+                size += serializedNullableSize(read.importMetadata, version, ImportMetadata.serializer);
+            }
+            else if (read.items.length > 0)
             {
                 size += serializedArraySize(read.items, tablesAndKeys, version, TxnNamedRead.serializer);
                 size += serializedNullableSize(read.cassandraConsistencyLevel, consistencyLevelSerializer);

src/java/org/apache/cassandra/streaming/StreamOperation.java

@@ -26,7 +26,8 @@ public enum StreamOperation
     BOOTSTRAP("Bootstrap", false, true, false),
     REBUILD("Rebuild", false, true, false),
     BULK_LOAD("Bulk Load", true, false, false),
-    REPAIR("Repair", true, false, true);
+    REPAIR("Repair", true, false, true),
+    ACCORD_SSTABLE_IMPORT("Accord SSTable Import", false, false, false);
 
     private final String description;
     private final boolean requiresViewBuild;

test/distributed/org/apache/cassandra/distributed/impl/Instance.java

@@ -139,6 +139,7 @@ import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.service.StorageServiceMBean;
 import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.service.accord.debug.AccordRemoteTracing;
+import org.apache.cassandra.service.accord.LocalTransfers;
 import org.apache.cassandra.service.paxos.PaxosRepair;
 import org.apache.cassandra.service.paxos.PaxosState;
 import org.apache.cassandra.service.paxos.uncommitted.UncommittedTableData;
@@ -1044,7 +1045,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
                                 SnapshotManager.instance::close,
                                 () -> IndexStatusManager.instance.shutdownAndWait(1L, MINUTES),
                                 DiskErrorsHandlerService::close,
-                                () -> ThreadLocalMetrics.shutdownCleaner(1L, MINUTES)
+                                () -> ThreadLocalMetrics.shutdownCleaner(1L, MINUTES),
+                                () -> LocalTransfers.instance.shutdownNowAndWait(1L, MINUTES)
             );
 
             internodeMessagingStarted = false;

test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java

@@ -0,0 +1,637 @@
+/*
+ * 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.nio.file.Files;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+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;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import net.bytebuddy.implementation.bind.annotation.This;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.streaming.CassandraStreamReceiver;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInstanceInitializer;
+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.distributed.test.sai.SAIUtil;
+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;
+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.jupiter.api.Assertions.assertTrue;
+
+public class AccordImportSSTableTest extends TestBaseImpl
+{
+    private static final String TABLE = "tbl";
+    private static final String KEYSPACE_TABLE = String.format("%s.%s", KEYSPACE, TABLE);
+    private static final String TABLE_SCHEMA_CQL = String.format(withKeyspace("CREATE TABLE %s." + TABLE + " (k int primary key, v int);"));
+
+    @Test
+    public void testImportSSTables() throws Throwable
+    {
+        String file = writeSSTables(new int[] { 1, 2 }, new int[] { 3 });
+
+        try (Cluster cluster = init(builder().withNodes(3)
+                                             .withoutVNodes()
+                                             .withDataDirCount(1)
+                                             .withConfig((config) ->
+                                                         config
+                                                         .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            createSchema(cluster);
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+                cfs.importNewSSTables(paths, true, true, true, true, true, true, true);
+            });
+
+            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();
+            });
+        }
+    }
+
+    @Test
+    public void testImportSSTablesBuildsIndex() throws Throwable
+    {
+        String file = writeSSTables(new int[] { 1 });
+
+        try (Cluster cluster = init(builder().withNodes(3)
+                                             .withoutVNodes()
+                                             .withDataDirCount(1)
+                                             .withConfig((config) ->
+                                                         config
+                                                         .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            createSchema(cluster);
+
+            String indexName = "v_idx";
+            cluster.schemaChange(withKeyspace("CREATE INDEX " + indexName + " ON " + KEYSPACE_TABLE + " (v) USING 'sai'"));
+
+            long mark = cluster.get(1).logs().mark();
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+                cfs.importNewSSTables(paths, true, true, true, true, true, true, true);
+            });
+
+            List<String> logs = cluster.get(1).logs().watchFor(mark, Duration.ofMinutes(1), "Submitting incremental index build of " + indexName).getResult();
+            Assertions.assertThat(logs).isNotEmpty();
+
+            Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
+
+            SAIUtil.assertIndexQueryable(cluster, KEYSPACE, indexName);
+
+            cluster.forEach(instance -> {
+                Object[][] rows = instance.executeInternal(withKeyspace("SELECT * FROM " + KEYSPACE_TABLE + " WHERE v = 1"));
+                assertRows(rows, row(1, 1));
+            });
+        }
+    }
+
+    @Test
+    public void testSSTableImportWithConcurrentTopologyChangeFails() throws Throwable
+    {
+        String file = writeSSTables(new int[] { 1 });
+
+        try (Cluster cluster = init(builder().withNodes(3)
+                                             .withoutVNodes()
+                                             .withDataDirCount(1)
+                                             .withInstanceInitializer(ByteBuddyInjections.StallImportTxn.install(1))
+                                             .withConfig((config) ->
+                                                         config
+                                                         .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            createSchema(cluster);
+
+            Thread importer = new Thread(() -> {
+                cluster.get(1).runOnInstance(() -> {
+                    ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                    Set<String> paths = Set.of(file);
+                    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();
+                });
+            }, "importer");
+
+            importer.start();
+
+            cluster.get(1).runOnInstance(() -> {
+                StorageService.instance.move(Long.toString(Long.parseLong(getOnlyElement(StorageService.instance.getTokens())) + 1));
+                State.waitForTopologyChange.countDown();
+            });
+
+            importer.join();
+
+            Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
+
+            assertSSTableCount(cluster, 0);
+        }
+    }
+
+    @Test
+    public void testSSTableImportStreamingFailedCleanup() throws Throwable
+    {
+        String file = writeSSTables(new int[] { 1, 2, 3 });
+
+        int FAILED_STREAM = 3;
+        try (Cluster cluster = init(builder().withNodes(3)
+                                             .withoutVNodes()
+                                             .withDataDirCount(1)
+                                             .withInstanceInitializer(ByteBuddyInjections.FailIncomingStream.install(FAILED_STREAM))
+                                             .withConfig((config) ->
+                                                         config
+                                                         .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            createSchema(cluster);
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+                Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true));
+            });
+
+            // We exclude the missed instance because the SSTables streamed to the pending directory
+            // are not linked to LocalTransfers and hence cleanup does not know which SSTables to clean up
+            assertPendingDirs(cluster.stream().filter(instance -> instance != cluster.get(FAILED_STREAM)).collect(Collectors.toList()), (File pendingUuidDir) -> {
+                Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty();
+            });
+
+            assertLocalSelect(cluster, rows -> assertRows(rows, EMPTY_ROWS));
+        }
+    }
+
+    @Test
+    public void testSSTableImportBounceAfterPending() throws Throwable
+    {
+        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
+        try (Cluster cluster = init(builder().withNodes(3).withoutVNodes()
+                                             .withDataDirCount(1).withConfig((config) ->
+                                                                             config
+                                                                             .set("accord.permit_local_delivery", false)
+                                                                             .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            createSchema(cluster);
+
+            // Prevent SSTables from being moved to the live set
+            cluster.filters().outbound().verbs(Verb.ACCORD_STABLE_THEN_READ_REQ.id).drop();
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+                Assertions.assertThatThrownBy(() -> cfs.importNewSSTables(paths, true, true, true, true, true, true, true));
+            });
+
+            assertLocalSelect(cluster, rows -> assertRows(rows, EMPTY_ROWS));
+
+            bounce(cluster);
+
+            assertLocalSelect(cluster, rows -> assertRows(rows, EMPTY_ROWS));
+        }
+    }
+
+    @Test
+    public void testRecoveryCoordinatorPerformsImport() throws Throwable
+    {
+        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
+        try (Cluster cluster = init(builder().withNodes(3).withoutVNodes()
+                                             .withDataDirCount(1).withConfig((config) ->
+                                                                             config
+                                                                             .set("accord.recover_txn", "100ms")
+                                                                             .set("accord.permit_local_delivery", false)
+                                                                             .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            createSchema(cluster);
+
+            // Simulate a network partition right before the StableThenRead message is sent
+            // so that a recovery coordinator can pick up the ImportTxn
+            cluster.filters().outbound().messagesMatching((from, to, msg) -> {
+                if (from == 1 && msg.verb() == Verb.ACCORD_STABLE_THEN_READ_REQ.id)
+                {
+                    cluster.filters().outbound().from(1).drop();
+                    return true;
+                }
+                return false;
+            }).drop();
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+
+                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");
+            });
+
+            Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS);
+
+            assertSSTableCount(cluster, 1);
+
+            assertLocalSelect(cluster, rows -> { assertRows(rows, row(1, 1), row(2, 1), row(3, 1)); });
+        }
+    }
+
+    @Test
+    public void testRecoveryCoordinatorPerformsImport2() throws Throwable
+    {
+        String file = writeSSTables(new int[] { 1, 2, 3 });
+
+        try (Cluster cluster = init(builder().withNodes(3).withoutVNodes()
+                                             .withDataDirCount(1).withConfig((config) ->
+                                                                             config
+                                                                             .set("accord.recover_txn", "100ms")
+                                                                             .set("accord.permit_local_delivery", false)
+                                                                             .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            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
+            cluster.filters().outbound().messagesMatching((from, to, msg) -> {
+                if (from == 1 && msg.verb() == Verb.ACCORD_STABLE_THEN_READ_REQ.id)
+                {
+                    // We prevent nodes 1 & 3 from receiving the StableThenRead message,
+                    // from node 1 and then prevent node 1 from receiving any more messages
+                    cluster.filters().outbound().from(1).to(1, 3).drop();
+                    cluster.filters().inbound().to(1).drop();
+
+                    // We still want node 2 to receive the message, so the ImportTxn is
+                    // stable, however once that is done we do not want to receive any more messages
+                    // from node 1
+                    if (to == 2)
+                    {
+                        cluster.filters().outbound().from(1).drop();
+                        return false;
+                    }
+                    return true;
+                }
+                return false;
+            }).drop();
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+                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");
+            });
+
+            Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS);
+
+            Iterable<IInvokableInstance> up = cluster.stream()
+                                                     .filter(instance -> instance != cluster.get(1))
+                                                     .collect(Collectors.toList());
+
+            assertSSTableCount(up, 1);
+
+            assertLocalSelect(up, rows -> { assertRows(rows, row(1, 1), row(2, 1), row(3, 1)); });
+        }
+    }
+
+    @Test
+    public void testImportSSTablesCleanupWithMultipleDataDirectories() throws Throwable
+    {
+        String file = writeSSTables(new int[] { 1, 2 }, new int[] { 3 });
+
+        try (Cluster cluster = init(builder().withNodes(3)
+                                             .withoutVNodes()
+                                             .withDataDirCount(3)
+                                             .withConfig((config) ->
+                                                         config
+                                                         .with(Feature.NETWORK, Feature.GOSSIP)).start()))
+        {
+            createSchema(cluster);
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+                cfs.importNewSSTables(paths, true, true, true, true, true, true, true);
+            });
+
+            Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
+
+            assertSSTableCount(cluster, 2);
+
+            assertLocalSelect(cluster, rows -> { assertRows(rows, row(1, 1), row(2, 1), row(3, 1)); });
+
+            assertPendingDirs(cluster, (File pendingUuidDir) -> {
+                Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty();
+            });
+        }
+    }
+
+    @Test
+    public void testImportSSTableFailsActivation() throws Throwable
+    {
+        String file = writeSSTables(new int[] { 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()))
+        {
+            createSchema(cluster);
+
+            cluster.get(1).runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Set<String> paths = Set.of(file);
+                cfs.importNewSSTables(paths, true, true, true, true, true, true, true);
+            });
+
+            Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS);
+
+            assertSSTableCount(cluster, 1);
+
+            assertLocalSelect(cluster, rows -> { assertRows(rows, row(1, 1)); });
+        }
+    }
+
+    @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(15, 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 -> {
+            try
+            {
+                instance.shutdown().get();
+            }
+            catch (InterruptedException | ExecutionException e)
+            {
+                throw new RuntimeException(e);
+            }
+            instance.startup();
+        });
+    }
+
+    private static void assertLocalTransferIsCleanup(Iterable<IInvokableInstance> validate)
+    {
+        for (IInvokableInstance instance : validate)
+        {
+            instance.runOnInstance(() -> {
+                assertTrue(LocalTransfers.instance.local.isEmpty());
+                assertTrue(LocalTransfers.instance.coordinating.isEmpty());
+            });
+        }
+    }
+
+    private static void assertPendingDirs(Iterable<IInvokableInstance> validate, IIsolatedExecutor.SerializableConsumer<File> forPendingUuidDir)
+    {
+        for (IInvokableInstance instance : validate)
+        {
+            instance.runOnInstance(() -> {
+                Set<File> allPendingDirs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE).getDirectories().getPendingLocations();
+                for (File pendingDir : allPendingDirs)
+                {
+                    File[] pendingUuidDirs = pendingDir.listUnchecked(File::isDirectory);
+                    for (File pendingUuidDir : pendingUuidDirs)
+                    {
+                        forPendingUuidDir.accept(pendingUuidDir);
+                    }
+                }
+            });
+        }
+    }
+
+    private static void assertSSTableCount(Iterable<IInvokableInstance> validate, int count)
+    {
+        for (IInvokableInstance instance : validate)
+        {
+            instance.runOnInstance(() -> {
+                ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+                Assertions.assertThat(cfs.getLiveSSTables().size()).isEqualTo(count);
+            });
+        }
+    }
+
+    private static void assertLocalSelect(Iterable<IInvokableInstance> validate, IIsolatedExecutor.SerializableConsumer<Object[][]> onRows)
+    {
+        for (IInvokableInstance instance : validate)
+        {
+            {
+                Object[][] rows = instance.executeInternal(withKeyspace("SELECT * FROM %s." + TABLE));
+                onRows.accept(rows);
+            }
+        }
+    }
+
+    @Shared
+    public static class State
+    {
+        public static CountDownLatch waitForTopologyChange = new CountDownLatch(1);
+        public static AtomicBoolean shouldFailDisk = new AtomicBoolean(true);
+    }
+
+    public static class ByteBuddyInjections
+    {
+        public static class StallImportTxn
+        {
+            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(CoordinatedTransfer.class)
+                                           .method(named("performImportTxn"))
+                                           .intercept(MethodDelegation.to(ByteBuddyInjections.StallImportTxn.class))
+                                           .make()
+                                           .load(cl, ClassLoadingStrategy.Default.INJECTION);
+                };
+            }
+
+            @SuppressWarnings("unused")
+            public static void performImportTxn(@SuperCall Callable<Void> r) throws Exception
+            {
+                State.waitForTopologyChange.await();
+                r.call();
+            }
+        }
+
+        public static class FailIncomingStream
+        {
+            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(CassandraStreamReceiver.class)
+                                           .method(named("finished").and(takesNoArguments()))
+                                           .intercept(MethodDelegation.to(FailIncomingStream.class))
+                                           .make()
+                                           .load(cl, ClassLoadingStrategy.Default.INJECTION);
+                };
+            }
+
+            @SuppressWarnings("unused")
+            public static void finished(@This CassandraStreamReceiver self)
+            {
+                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<Component> components, boolean copyData, @SuperCall Callable<SSTableReader> r) throws Exception
+            {
+                if (State.shouldFailDisk.get())
+                {
+                    State.shouldFailDisk.set(false);
+                    throw new RuntimeException("Failing move and open SSTable");
+                }
+                return r.call();
+            }
+        }
+    }
+}

test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java

@@ -712,7 +712,7 @@ public class RouteIndexTest extends CQLTester
                        .getColumnFamilyStore(AccordKeyspace.JOURNAL);
     }
 
-    private static Gen<TokenRange> rangeGen(RandomSource rand, List<TableId> tables)
+    public static Gen<TokenRange> rangeGen(RandomSource rand, List<TableId> tables)
     {
         Gen.IntGen tokenGen = TOKEN_DISTRIBUTION.next(rand);
         Gen<TableId> tableIdGen = Gens.mixedDistribution(tables).next(rand);

test/unit/org/apache/cassandra/service/accord/txn/TxnReadTest.java

@@ -0,0 +1,87 @@
+/*
+ * 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.service.accord.txn;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.io.Serializers;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.serializers.TableMetadatas;
+import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys;
+import org.apache.cassandra.service.accord.serializers.Version;
+import org.apache.cassandra.service.accord.txn.TxnRead.ImportMetadata;
+import org.apache.cassandra.utils.Generators;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static accord.utils.Property.qt;
+import static org.apache.cassandra.index.accord.RouteIndexTest.rangeGen;
+import static org.apache.cassandra.utils.CassandraGenerators.TABLE_METADATA_GEN;
+import static org.apache.cassandra.utils.Generators.toGen;
+
+public class TxnReadTest
+{
+    static
+    {
+        DatabaseDescriptor.clientInitialization();
+        DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+    }
+
+    @Test
+    public void importMetaDataSerde()
+    {
+        DataOutputBuffer output = new DataOutputBuffer();
+        qt().check(rs -> {
+            ImportMetadata importMetadata = new ImportMetadata(toGen(Generators.UUID_RANDOM_GEN).next(rs), toGen(Generators.timeUUID()).next(rs), rs.nextLong());
+            Serializers.testSerde(output, ImportMetadata.serializer, importMetadata, Version.LATEST);
+        });
+    }
+
+    @Test
+    public void importReadSerde()
+    {
+        DataOutputBuffer output = new DataOutputBuffer();
+        qt().check(rs -> {
+            int length = rs.nextInt(1, 10);
+            List<TableMetadata> tables = new ArrayList<>(length);
+            for (int i = 0; i < length; i++)
+                tables.add(toGen(TABLE_METADATA_GEN).next(rs));
+            TableMetadatas tableMetadatas = TableMetadatas.of(tables);
+            UUID uuid = toGen(Generators.UUID_RANDOM_GEN).next(rs);
+            TimeUUID timeUUID = toGen(Generators.timeUUID()).next(rs);
+            long epoch = rs.nextLong();
+            TokenRange range = rangeGen(rs, tables.stream().map(TableMetadata::id).collect(Collectors.toList())).next(rs);
+            TxnRead txnRead = TxnRead.createImport(tableMetadatas, range, uuid, timeUUID, epoch);
+            Serializers.testSerde(output, TxnRead.serializer, txnRead, tablesAndKeys(txnRead), Version.LATEST);
+        });
+    }
+
+    private static TableMetadatasAndKeys tablesAndKeys(TxnRead read)
+    {
+        return new TableMetadatasAndKeys(read.tables, read.keys());
+    }
+}