diff --git a/.gitmodules b/.gitmodules index 616dacf610..92d672139e 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "modules/accord"] path = modules/accord - url = https://github.com/apache/cassandra-accord.git - branch = trunk + url = https://github.com/alanwang67/cassandra-accord.git + branch = CASSANDRA-20595 diff --git a/modules/accord b/modules/accord index fff32de2e9..e2352bc34e 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit fff32de2e915772fbc70d16b1c32346313877838 +Subproject commit e2352bc34edaf9e8662068ba163519c7ba2a50fb diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index db15db6729..568ac89230 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/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 sstables, Consumer onRowCacheInvalidation, Consumer onCounterCacheInvalidation) + { + if (isRowCacheEnabled() || metadata().isCounter()) + { + List> boundsToInvalidate = new ArrayList<>(sstables.size()); + sstables.forEach(sstable -> boundsToInvalidate.add(new Bounds<>(sstable.getFirst().getToken(), sstable.getLast().getToken()))); + Set> 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> boundsToInvalidate) { int invalidatedKeys = 0; diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java index 236b7186b2..a567ec89d2 100644 --- a/src/java/org/apache/cassandra/db/Directories.java +++ b/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 - Path dirPath = dir.toAbsolute().toPath(); - Path locationPath = dataDirectory.location.toAbsolute().toPath(); - if (dirPath.startsWith(locationPath)) + if (dataDirectory.contains(dir)) return dir; } return null; @@ -726,6 +725,33 @@ public class Directories return new File(snapshotDir, "schema.cql"); } + @VisibleForTesting + public Set getPendingLocations() + { + Set 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(); diff --git a/src/java/org/apache/cassandra/db/SSTableImporter.java b/src/java/org/apache/cassandra/db/SSTableImporter.java index c6361d6c72..2ae728a776 100644 --- a/src/java/org/apache/cassandra/db/SSTableImporter.java +++ b/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> listers = getSSTableListers(options.srcPaths); Set 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); - throw new RuntimeException("Failed adding SSTables", t); + if (isAccordEnabled) + { + String msg = "Failed adding SSTables on local node; note the import may still have been committed by a recovery coordinator"; + throw new RuntimeException(msg, t); + } + else + { + logger.error("[{}] Failed adding SSTables", importID, t); + throw new RuntimeException("Failed adding SSTables", t); + } } logger.info("[{}] Done loading load new SSTables for {}/{}", importID, cfs.getKeyspaceName(), cfs.getTableName()); diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java index f2f1784a9c..61cd590891 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java +++ b/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) diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java index 9a6836d34a..c1c59fc0ea 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java +++ b/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); } diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java index 1bbc3d3507..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; @@ -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 diff --git a/src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java new file mode 100644 index 0000000000..953f800ac5 --- /dev/null +++ b/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); + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java index 1db8ef679d..e5004ffe2e 100644 --- a/src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java +++ b/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 boundaries; - private final List directories; + protected final List 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 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; } diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index f0721fe997..d746ebbc63 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/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 ), diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 111e207474..7610030934 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/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 sstables, TableMetadata metadata) + { + logger.info("Creating Accord bulk transfer for keyspace '{}' table '{}' SSTables {}...", keyspace, metadata.name, sstables); + + Topology topology = topology().current(); + + Map 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); + } } diff --git a/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java b/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java new file mode 100644 index 0000000000..18371b53e2 --- /dev/null +++ b/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 nodeStreamingContext; + SingleTransferResult streamResult = SingleTransferResult.Init(); + + public CoordinatedTransfer(UUID importID, TableMetadata tableMetadata, Map 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 accordResult = AccordService.instance().coordinateAsync(tableMetadata.epoch.getEpoch(), txn, ConsistencyLevel.ALL, Dispatcher.RequestTime.forImmediateExecution()); + accordResult.awaitAndGet(); + } + + private void stream() + { + Future 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 entry : nodeStreamingContext.entrySet()) + { + InetAddressAndPort to = entry.getKey(); + NodeStreamingMetadata sstablesForNode = entry.getValue(); + List> ranges = sstablesForNode.ranges; + + for (Map.Entry> positionsForSSTables : sstablesForNode.positionsForSSTables.entrySet()) + { + SSTableReader sstable = positionsForSSTables.getKey(); + List 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 implements RequestCallbackWithFailure + { + final Set responses = ConcurrentHashMap.newKeySet(nodeStreamingContext.size()); + + @Override + public void onResponse(Message 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> 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> 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> ranges; + final Map> positionsForSSTables; + + public NodeStreamingMetadata(Node.Id id, Map> positionsForSSTables, List> 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 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 getNodeStreamingContext(Collection sstables, Topology topology, AccordEndpointMapper endpointMapper) + { + Map 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> 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> positionsForSSTables = new HashMap<>(); + + for (SSTableReader sstable : sstables) + { + List 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; + } +} diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index 77296264e0..eb4098087b 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/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 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 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 sstables, TableMetadata metadata) + { + delegate.executeTransfer(importID, keyspace, sstables, metadata); + } + + @Override + public void receivedSSTableImport(PendingLocalTransfer transfer) + { + delegate.receivedSSTableImport(transfer); + } + @Override public void ensureMinHlc(long minHlc) { diff --git a/src/java/org/apache/cassandra/service/accord/LocalTransfers.java b/src/java/org/apache/cassandra/service/accord/LocalTransfers.java new file mode 100644 index 0000000000..310316e6df --- /dev/null +++ b/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 coordinating = new ConcurrentHashMap<>(); + + // Added when we have a streamed SSTable in our pending directory + public final Map 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// + if (!transfer.sstables.isEmpty()) + { + Set 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 verbHandler = message -> { + LocalTransfers.instance().purge(message.payload.planId); + MessagingService.instance().respond(NoPayload.noPayload, message); + }; +} diff --git a/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java b/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java new file mode 100644 index 0000000000..c7f03fe996 --- /dev/null +++ b/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 sstables; + final long createdAt = currentTimeMillis(); + transient String keyspace; + @Nullable CountDownLatch latch; + + volatile boolean activated = false; + + public PendingLocalTransfer(TableId tableId, TimeUUID planId, Collection 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 moved = new ArrayList<>(sstables.size()); + Collection 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 onRowCacheInvalidation = invalidatedKeys -> { + logger.debug("{} Invalidated {} row cache entries on table {}.{} after activating transfer", + logPrefix(), invalidatedKeys, cfs.getKeyspaceName(), cfs.getTableName()); + }; + Consumer 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); + } +} diff --git a/src/java/org/apache/cassandra/service/accord/TransferFailed.java b/src/java/org/apache/cassandra/service/accord/TransferFailed.java new file mode 100644 index 0000000000..0de6555aa2 --- /dev/null +++ b/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 serializer = new IVersionedSerializer() + { + @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 + + '}'; + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java index 29f0220790..5905fc1b24 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java +++ b/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 range, TableId tableId) { // Should already have been unwrapped @@ -402,6 +410,15 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized performSSTableImport(AccordExecutor executor, TxnRead.ImportMetadata importMetadata, Timestamp executeAt) + { + Callable callable = () -> { + LocalTransfers.instance.activatePendingTransfers(importMetadata, executeAt.epoch()); + return new TxnData(); + }; + return submit(executor, callable, callable); + } + private AsyncChain submit(AccordExecutor executor, Callable readCallable, Object describe) { return executor.buildDebuggable(readCallable, describe); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java index bddb2b6fba..6537612097 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java +++ b/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 implements Read { - private static final TxnRead EMPTY_KEY = new TxnRead(TableMetadatas.none(), Domain.Key); - private static final TxnRead EMPTY_RANGE = new TxnRead(TableMetadatas.none(), Domain.Range); + 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, null); private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY_KEY); private static final Comparator 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 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 items, @Nullable ConsistencyLevel cassandraConsistencyLevel) + private TxnRead(TableMetadatas tables, @Nonnull List 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 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 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 readCommands, ConsistencyLevel consistencyLevel, TableMetadatasAndKeys.KeyCollector keyCollector) @@ -163,13 +185,13 @@ public class TxnRead extends AbstractKeySorted 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 implements Read List 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 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 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 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 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 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 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 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 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 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 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); diff --git a/src/java/org/apache/cassandra/streaming/StreamOperation.java b/src/java/org/apache/cassandra/streaming/StreamOperation.java index b1c5908f7f..4b87dc327b 100644 --- a/src/java/org/apache/cassandra/streaming/StreamOperation.java +++ b/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; diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index c3b0e3ebd4..9b42806dd8 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/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; diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java new file mode 100644 index 0000000000..580a80f782 --- /dev/null +++ b/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 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 paths = Set.of(file); + cfs.importNewSSTables(paths, true, true, true, true, true, true, true); + }); + + List 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 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 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 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 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 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 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 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 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 validate) + { + for (IInvokableInstance instance : validate) + { + instance.runOnInstance(() -> { + assertTrue(LocalTransfers.instance.local.isEmpty()); + assertTrue(LocalTransfers.instance.coordinating.isEmpty()); + }); + } + } + + private static void assertPendingDirs(Iterable validate, IIsolatedExecutor.SerializableConsumer forPendingUuidDir) + { + for (IInvokableInstance instance : validate) + { + instance.runOnInstance(() -> { + Set 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 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 validate, IIsolatedExecutor.SerializableConsumer 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 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 components, boolean copyData, @SuperCall Callable r) throws Exception + { + if (State.shouldFailDisk.get()) + { + State.shouldFailDisk.set(false); + throw new RuntimeException("Failing move and open SSTable"); + } + return r.call(); + } + } + } +} 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()); + } +}