tests + fix

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;
@@ -245,6 +246,7 @@ public class CassandraStreamReceiver implements StreamReceiver
         {
             if (requiresWritePath)
             {
+                Invariants.require(session.streamOperation() != StreamOperation.ACCORD_SSTABLE_IMPORT);
                 sendThroughWritePath(cfs, readers);
             }
             else

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;
@@ -118,22 +119,22 @@ public class CoordinatedTransfer
     {
         logger.debug("{} Executing Accord bulk transfer {}", logPrefix(), this);
         LocalTransfers.instance().save(this);
-
+        PendingLocalTransfer pendingLocalTransfer = LocalTransfers.instance().local.get(streamResult.planId);
+        CountDownLatch latch = new CountDownLatch(1);
+        pendingLocalTransfer.registerLatch(latch);
         stream();
 
         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");
             }
-
-            Invariants.require(LocalTransfers.instance.local.get(streamResult.planId()) == null);
-            LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this);
         }
-        catch (ReadTimeoutException 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

@@ -21,13 +21,18 @@ 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;
@@ -52,6 +57,7 @@ public class PendingLocalTransfer
     final Collection<SSTableReader> sstables;
     final long createdAt = currentTimeMillis();
     transient String keyspace;
+    @Nullable CountDownLatch latch;
 
     volatile boolean activated = false;
 
@@ -64,6 +70,12 @@ public class PendingLocalTransfer
         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.
      */
@@ -81,7 +93,11 @@ public class PendingLocalTransfer
                         logPrefix(), metadata.getStreamingEpoch(), executeAtEpoch);
 
             if (isCoordinator)
+            {
+                Invariants.require(latch != null);
+                latch.countDown();
                 coordinatedTransfer.importTxnEpochMismatch = true;
+            }
 
             LocalTransfers.instance().schedulePendingLocalTransferCleanup(planId);
 

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());
+    }
+}