Merge d5a8c998b0 into 10557d7ffe

modules/accord

@@ -1 +1 @@
-Subproject commit e2352bc34edaf9e8662068ba163519c7ba2a50fb
+Subproject commit c12dd6f8767dccd0c7cd3480f938845c9e24cb94

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

@@ -27,6 +27,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -97,6 +98,8 @@ public class CoordinatedTransfer
 
     final Map<InetAddressAndPort, NodeStreamingMetadata> nodeStreamingContext;
     SingleTransferResult streamResult = SingleTransferResult.Init();
+    // If importTxnEpochMismatch is true, all replicas will deterministically not import the SSTables in the pending directory
+    volatile boolean importTxnEpochMismatch = false;
 
     public CoordinatedTransfer(UUID importID, TableMetadata tableMetadata, Map<InetAddressAndPort, NodeStreamingMetadata> nodeStreamingContext, long streamingEpoch, TokenRange allSSTableRanges)
     {
@@ -117,13 +120,21 @@ public class CoordinatedTransfer
         logger.debug("{} Executing Accord bulk transfer {}", logPrefix(), this);
         LocalTransfers.instance().save(this);
         stream();
+        PendingLocalTransfer pendingLocalTransfer = LocalTransfers.instance().local.get(streamResult.planId);
+        CountDownLatch latch = new CountDownLatch(1);
+        pendingLocalTransfer.registerLatch(latch);
 
         try
         {
             performImportTxn();
-
+            latch.await();
+            if (importTxnEpochMismatch)
+            {
+                LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this);
+                throw new RuntimeException("SSTable import failed because of a concurrent topology change; please retry the operation");
+            }
         }
-        catch (Exception e)
+        catch (ReadTimeoutException | InterruptedException e)
         {
             throw new RuntimeException("SSTable import failed locally; however the operation may still be applied by the recovery coordinator", e);
         }

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

@@ -84,7 +84,27 @@ public class PendingLocalTransfer
         if (activated)
             return;
 
-        Invariants.require(metadata.getStreamingEpoch() == executeAtEpoch);
+        CoordinatedTransfer coordinatedTransfer = LocalTransfers.instance.coordinating.get(metadata.getImportID());
+        boolean isCoordinator = coordinatedTransfer != null;
+
+        if (metadata.getStreamingEpoch() != executeAtEpoch)
+        {
+            logger.info("{} Failing activation of pending SSTables because streaming epoch {} != importTxn executeAt epoch {}",
+                        logPrefix(), metadata.getStreamingEpoch(), executeAtEpoch);
+
+            if (isCoordinator)
+            {
+                Invariants.require(latch != null);
+                latch.countDown();
+                coordinatedTransfer.importTxnEpochMismatch = true;
+            }
+
+            LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId);
+
+            activated = true;
+            return;
+        }
+
         long startedActivation = currentTimeMillis();
         logger.info("{} Activating transfer {}, {} ms since pending", logPrefix(), this, startedActivation - createdAt);
         ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(tableId);

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

@@ -535,14 +535,6 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
         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

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

@@ -170,7 +170,10 @@ public class AccordImportSSTableTest extends TestBaseImpl
                     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();
+                              .cause()
+                              .isInstanceOf(RuntimeException.class)
+                              .hasMessageContaining("SSTable import failed because of a concurrent topology change; please retry the operation");
+
                 });
             }, "importer");
 
@@ -450,7 +453,7 @@ public class AccordImportSSTableTest extends TestBaseImpl
                 cfs.importNewSSTables(Set.of(file), true, true, true, true, true, true, true);
             });
 
-            Uninterruptibles.sleepUninterruptibly(15, TimeUnit.SECONDS);
+            Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS);
 
             assertLocalSelect(cluster, rows -> assertRows(rows, row(1, 1), row(2, 1), row(3, 1)));
             assertSSTableCount(cluster, 1);