diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java index 0a0ac311b3..f950913c3f 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java +++ b/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 diff --git a/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java b/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java index 20ec86bc94..dc7b8a592b 100644 --- a/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java +++ b/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); } diff --git a/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java b/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java index 233c189d28..1fcf16ed05 100644 --- a/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java +++ b/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 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); diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java index 2e3a01d764..06d0d6361d 100644 --- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java @@ -712,7 +712,7 @@ public class RouteIndexTest extends CQLTester .getColumnFamilyStore(AccordKeyspace.JOURNAL); } - private static Gen rangeGen(RandomSource rand, List tables) + public static Gen rangeGen(RandomSource rand, List tables) { Gen.IntGen tokenGen = TOKEN_DISTRIBUTION.next(rand); Gen tableIdGen = Gens.mixedDistribution(tables).next(rand); diff --git a/test/unit/org/apache/cassandra/service/accord/txn/TxnReadTest.java b/test/unit/org/apache/cassandra/service/accord/txn/TxnReadTest.java new file mode 100644 index 0000000000..5d23c66906 --- /dev/null +++ b/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 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()); + } +}