From c8079bd97f177655249a4197a951a7d53ba66071 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Fri, 17 Jul 2026 10:53:19 -0700 Subject: [PATCH] rebase on trunk --- .gitmodules | 4 +- modules/accord | 2 +- .../cassandra/db/ColumnFamilyStore.java | 27 +- .../org/apache/cassandra/db/Directories.java | 41 +- .../apache/cassandra/db/SSTableImporter.java | 14 +- .../CassandraEntireSSTableStreamReader.java | 4 + .../db/streaming/CassandraStreamReader.java | 12 +- .../db/streaming/CassandraStreamReceiver.java | 13 + .../RangeAwarePendingSSTableWriter.java | 44 ++ .../io/sstable/RangeAwareSSTableWriter.java | 15 +- src/java/org/apache/cassandra/net/Verb.java | 5 + .../service/accord/AccordService.java | 25 + .../service/accord/CoordinatedTransfer.java | 499 ++++++++++++++++++ .../service/accord/IAccordService.java | 31 ++ .../service/accord/LocalTransfers.java | 253 +++++++++ .../service/accord/PendingLocalTransfer.java | 132 +++++ .../service/accord/TransferFailed.java | 72 +++ .../service/accord/txn/TxnNamedRead.java | 19 + .../cassandra/service/accord/txn/TxnRead.java | 168 +++++- .../cassandra/streaming/StreamOperation.java | 3 +- .../cassandra/distributed/impl/Instance.java | 4 +- .../test/accord/AccordImportSSTableTest.java | 484 +++++++++++++++++ 22 files changed, 1842 insertions(+), 29 deletions(-) create mode 100644 src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java create mode 100644 src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java create mode 100644 src/java/org/apache/cassandra/service/accord/LocalTransfers.java create mode 100644 src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java create mode 100644 src/java/org/apache/cassandra/service/accord/TransferFailed.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java 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..53aa05ffcb 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit fff32de2e915772fbc70d16b1c32346313877838 +Subproject commit 53aa05ffcbcf994d25d9239d4a71c493b6c8e98d 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..587b8642ca 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(cfs.keyspace.getName(), newSSTables, metadata); + else + cfs.getTracker().addSSTables(newSSTables); for (SSTableReader reader : newSSTables) { if (options.invalidateCaches && cfs.isRowCacheEnabled()) diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java index f2f1784a9c..e594fb399d 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java @@ -159,10 +159,14 @@ public class CassandraEntireSSTableStreamReader implements IStreamReader private File getDataDir(ColumnFamilyStore cfs, long totalSize) throws IOException { + boolean isAccordEnabled = cfs.metadata().isAccordEnabled(); Directories.DataDirectory localDir = cfs.getDirectories().getWriteableLocation(totalSize); if (localDir == null) throw new IOException(format("Insufficient disk space to store %s", prettyPrintMemory(totalSize))); + if (isAccordEnabled) + 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..0a0ac311b3 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java @@ -51,9 +51,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 +117,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) { @@ -257,6 +260,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 340760370c..3f7e3009f5 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -39,6 +39,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 +103,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 +193,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; @@ -1434,4 +1438,25 @@ public class AccordService implements IAccordService, Shutdownable { return journal.configuration(); } + + public void executeTransfer(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(node().nextCoordinationId(), keyspace, 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..f090f5cb3b --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java @@ -0,0 +1,499 @@ +/* + * 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.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +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.Keyspace; +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.RequestFailure; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.locator.AbstractReplicationStrategy; +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; + +/** + * Orchestrates the lifecycle of a bulk data transfer for Accord. Bulk data transfers only work on a single shard, + * where the current instance is coordinating the transfer. This means that the node that the bulk data transfer + * request is submitted to will only perform the transfer for the ranges of the SSTables that it owns. + *

+ * The transfer proceeds through the following two phases: + *

    + *
  1. + * Streaming + * The coordinator streams SSTables to all replicas in parallel. Replicas store received data in a "pending" + * location where it's persisted to disk but not yet visible to reads. Once all replicas have received + * their streams, the SSTables are activated using a two-phase + * commit protocol, making them part of the live set and visible to reads. + *
    2. + *
  2. + * ImportTxn + * Once all replicas have received their streams, the SSTables are activated by creating an ImportTxn. An + * ImportTxn is equivalent to a range read txn that atomically activates the SSTables making them part of the live set + * and visible to reads. Note that because this is done with a range read txn and not a range write txn, clients + * performing concurrent reads against any nodes involved in the SSTable import can see inconsistent reads for + * a window of time because read txns can occur concurrently with other read txns. However, there will + * not be any data inconsistency, because read txns can not be interleaved with other write txns. In cases where + * the coordinator fails when performing the ImportTxn, the recovery protocol for Accord txn's is performed and + * a recovery coordinator will push the import to completion. + *
    3. + *
+ * + * For simplicity, the coordinator streams to itself rather than using direct file copy. This ensures we can use the + * same lifecycle management for crash-safety and atomic add. + *

+ * In cases where there is a topology change that is concurrent with the ImportTxn. We do not import the SSTables, as + * the node with the streamed SSTables may no longer own those ranges and we rely on the client to reperform the import + * operation. + */ +public class CoordinatedTransfer +{ + private static final Logger logger = LoggerFactory.getLogger(CoordinatedTransfer.class); + + String logPrefix() + { + return String.format("[CoordinatedTransfer #%s]", id); + } + + private final Long id; + private final String keyspace; + private final TableMetadata tableMetadata; + private final long streamingEpoch; + private final TokenRange allSSTableRanges; + + final Map nodeStreamingContext; + final ConcurrentMap streamResults; + + public CoordinatedTransfer(Long id, String keyspace, TableMetadata tableMetadata, Map nodeStreamingContext, long streamingEpoch, TokenRange allSSTableRanges) + { + this.id = id; + this.keyspace = keyspace; + this.tableMetadata = tableMetadata; + this.nodeStreamingContext = nodeStreamingContext; + this.streamingEpoch = streamingEpoch; + this.allSSTableRanges = allSSTableRanges; + + this.streamResults = new ConcurrentHashMap<>(nodeStreamingContext.size()); + for (InetAddressAndPort ip: nodeStreamingContext.keySet()) + this.streamResults.put(ip, SingleTransferResult.Init()); + } + + public long id() + { + return id; + } + + void execute() + { + logger.debug("{} Executing Accord bulk transfer {}", logPrefix(), this); + LocalTransfers.instance().save(this); + + stream(); + + AbstractReplicationStrategy ars = Keyspace.open(keyspace).getReplicationStrategy(); + int blockFor = ConsistencyLevel.ALL.blockFor(ars); + int responses = 0; + for (Map.Entry entry : streamResults.entrySet()) + { + if (entry.getValue().state == SingleTransferResult.State.STREAM_COMPLETE) + responses++; + } + + Invariants.require(responses == blockFor); + performImportTxn(); + LocalTransfers.instance().scheduleCoordinatedTransferCleanup(this); + } + + private void performImportTxn() + { + TimeUUID[] planIds = streamResults.values().stream() + .filter(result -> result.planId != null) + .map(result -> result.planId) + .toArray(TimeUUID[]::new); + + TableMetadatas tables = TableMetadatas.of(tableMetadata); + TxnRead read = TxnRead.createImport(tables, allSSTableRanges, planIds, 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() + { + List> streaming = new ArrayList<>(streamResults.size()); + for (InetAddressAndPort to : streamResults.keySet()) + { + Future stream = LocalTransfers.instance().executor.submit(() -> { + SingleTransferResult result; + try + { + result = streamTask(to); + } + catch (StreamException | ExecutionException | InterruptedException | TimeoutException e) + { + Throwable cause = e instanceof ExecutionException ? e.getCause() : e; + markStreamFailure(to, cause); + throw Throwables.unchecked(cause); + } + + streamResults.put(to, result); + logger.info("{} Completed streaming to {}, {}", logPrefix(), to, this); + return null; + }); + streaming.add(stream); + } + + // Wait for all streams to complete, so we can clean up after failures. If we exit at the first failure, a + // future stream can complete. + LinkedList failures = null; + for (Future stream : streaming) + { + try + { + stream.get(); + } + catch (InterruptedException | ExecutionException e) + { + if (failures == null) + failures = new LinkedList<>(); + failures.add(e); + logger.error("{} Failed transfer due to", logPrefix(), e); + } + } + + if (failures != null && !failures.isEmpty()) + { + Throwable failure = failures.element(); + Throwable cause = failure instanceof ExecutionException ? failure.getCause() : failure; + maybeCleanupFailedStreams(cause); + + String msg = String.format("Failed streaming on %s instance(s): %s", failures.size(), failures); + throw new RuntimeException(msg, Throwables.unchecked(cause)); + } + + logger.info("{} All streaming completed successfully", logPrefix()); + } + + private SingleTransferResult streamTask(InetAddressAndPort to) 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); + + SSTablesForNode sstablesForNode = nodeStreamingContext.get(to); + List> ranges = nodeStreamingContext.get(to).ranges; + + for (Map.Entry> entry : sstablesForNode.positionsForSSTables.entrySet()) + { + SSTableReader sstable = entry.getKey(); + List positions = entry.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 {} to peer {}", logPrefix(), this, to); + StreamResultFuture execute = plan.execute(); + StreamState state; + try + { + state = execute.get(timeout, TimeUnit.MILLISECONDS); + logger.debug("{} Completed streaming transfer {} to peer {}", logPrefix(), this, to); + } + 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"); + + // If the SSTable doesn't contain any rows in the provided range, no streams delivered, nothing to activate + if (state.sessions().isEmpty()) + return SingleTransferResult.Noop(); + + return SingleTransferResult.StreamComplete(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().scheduleCoordinatedTransferCleanup(this); + } + catch (Throwable t) + { + if (cause != null) + t.addSuppressed(cause); + logger.error("{} Failed to notify peers of stream failure", logPrefix(), t); + } + } + + private void markStreamFailure(InetAddressAndPort to, Throwable cause) + { + TimeUUID planId; + if (cause instanceof StreamException) + planId = ((StreamException) cause).finalState.planId; + else + planId = null; + streamResults.computeIfPresent(to, (peer, result) -> result.streamFailed(planId)); + } + + private void notifyFailure() throws ExecutionException, InterruptedException + { + class NotifyFailure extends AsyncFuture implements RequestCallbackWithFailure + { + final Set responses = ConcurrentHashMap.newKeySet(streamResults.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(); + + for (Map.Entry entry : streamResults.entrySet()) + { + InetAddressAndPort to = entry.getKey(); + // Coordinator cleans up CoordinatedTransfer and PendingLocalTransfer separately, does not need to notify + if (FBUtilities.getBroadcastAddressAndPort().equals(to)) + continue; + + SingleTransferResult result = entry.getValue(); + if (result.planId == null) + { + logger.warn("{} Skipping notification of transfer failure to {} due to unknown planId", logPrefix(), to); + continue; + } + + logger.debug("{}, Notifying {} of transfer failure for plan {}", logPrefix(), to, result.planId); + notifyFailure.responses.add(to); + msgs.put(to, Message.out(Verb.TRACKED_TRANSFER_FAILED_REQ, new TransferFailed(result.planId))); + } + + for (Map.Entry> entry : msgs.entrySet()) + MessagingService.instance().sendWithCallback(entry.getValue(), entry.getKey(), notifyFailure); + + notifyFailure.get(); + } + + public static class SSTablesForNode + { + final Node.Id id; + final List> ranges; + final Map> positionsForSSTables; + + public SSTablesForNode(Node.Id id, Map> positionsForSSTables, List> ranges) + { + this.id = id; + this.positionsForSSTables = positionsForSSTables; + this.ranges = ranges; + } + } + + static class SingleTransferResult + { + enum State + { + INIT, + STREAM_NOOP, + 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); + } + + @VisibleForTesting + static SingleTransferResult StreamComplete(TimeUUID planId) + { + return new SingleTransferResult(State.STREAM_COMPLETE, planId); + } + + @VisibleForTesting + static SingleTransferResult Noop() + { + return new SingleTransferResult(State.STREAM_NOOP, null); + } + + public SingleTransferResult streamFailed(TimeUUID planId) + { + 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()) + { + Ranges rangesForNode = topology.rangesForNode(nodeId); + List> ranges = new ArrayList<>(); + for (accord.primitives.Range range : rangesForNode) + ranges.add(((TokenRange) range).toKeyspaceRange()); + + 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 SSTablesForNode(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..627a64cd4b 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -21,6 +21,7 @@ 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.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.BiConsumer; @@ -57,10 +58,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 +196,10 @@ public interface IAccordService Node node(); + void executeTransfer(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 +379,18 @@ public interface IAccordService return null; } + @Override + public void executeTransfer(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 +583,18 @@ public interface IAccordService return delegate.node(); } + @Override + public void executeTransfer(String keyspace, Set sstables, TableMetadata metadata) + { + delegate.executeTransfer(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..a899e6a166 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/LocalTransfers.java @@ -0,0 +1,253 @@ +/* + * 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.Map; +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 accord.utils.Invariants; + +import org.apache.cassandra.concurrent.ExecutorPlus; +import org.apache.cassandra.io.sstable.format.SSTableReader; +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.FBUtilities; +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 + private final Map coordinating = new ConcurrentHashMap<>(); + + // Added when we have a streamed SSTable in our pending directory + private 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.id(), 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 cleanupCoordinatedTransfer(CoordinatedTransfer transfer) + { + lock.writeLock().lock(); + try + { + purge(transfer); + } + finally + { + lock.writeLock().unlock(); + } + } + + private void cleanupPendingLocalTransfer(TimeUUID timeUUID) + { + lock.writeLock().lock(); + try + { + purge(local.get(timeUUID)); + } + finally + { + lock.writeLock().unlock(); + } + } + + private void purge(TransferFailed failed) + { + lock.writeLock().lock(); + try + { + PendingLocalTransfer pending = local.get(failed.planId); + if (pending == null) + { + logger.warn("Cannot purge unknown local pending transfer {}", failed); + return; + } + purge(pending); + } + finally + { + lock.writeLock().unlock(); + } + } + + private void purge(PendingLocalTransfer transfer) + { + logger.info("Cleaning up pending transfer {}", transfer); + + lock.writeLock().lock(); + try + { + // Delete the entire pending transfer directory /pending// + if (!transfer.sstables.isEmpty()) + { + SSTableReader sstable = transfer.sstables.iterator().next(); + File pendingDir = sstable.descriptor.directory; + + if (pendingDir.exists()) + { + Preconditions.checkState(pendingDir.absolutePath().contains(transfer.planId.toString())); + logger.debug("Deleting pending transfer directory: {}", pendingDir); + pendingDir.deleteRecursive(); + } + } + } + finally + { + lock.writeLock().unlock(); + } + } + + private void purge(CoordinatedTransfer transfer) + { + logger.info("Cleaning up completed coordinated transfer: {}", transfer); + + lock.writeLock().lock(); + try + { + coordinating.remove(transfer.id()); + + CoordinatedTransfer.SingleTransferResult localPending = transfer.streamResults.get(FBUtilities.getBroadcastAddressAndPort()); + PendingLocalTransfer localTransfer; + TimeUUID planId; + if (localPending != null && (planId = localPending.planId()) != null && (localTransfer = local.get(planId)) != null) + purge(localTransfer); + } + finally + { + lock.writeLock().unlock(); + } + } + + void scheduleCoordinatedTransferCleanup(CoordinatedTransfer transfer) + { + executor.submit(() -> { + try + { + cleanupCoordinatedTransfer(transfer); + } + catch (Throwable t) + { + logger.error("Cleanup failed", t); + } + }); + } + + void schedulePendingLocalTransferCleanup(TimeUUID timeUUID) + { + executor.submit(() -> { + try + { + cleanupPendingLocalTransfer(timeUUID); + } + catch (Throwable t) + { + logger.error("Cleanup failed", t); + } + }); + } + + public void activatePendingTransfers(TxnRead.ImportMetadata metadata) + { + lock.readLock().lock(); + try + { + int activatedTransfer = 0; + for (TimeUUID planId : metadata.getPlanIds()) + { + PendingLocalTransfer pendingLocalTransfer = local.get(planId); + if (pendingLocalTransfer != null) + { + activatedTransfer += 1; + pendingLocalTransfer.activate(); + } + } + + Invariants.require(activatedTransfer == 1); + } + 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); + 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..d402c5a2d8 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java @@ -0,0 +1,132 @@ +/* + * 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.function.Consumer; + +import com.google.common.base.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.ColumnFamilyStore; +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.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; + + 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(); + } + + /** + * Safely moves SSTables into the live set. + */ + public synchronized void activate() + { + if (activated) + return; + + 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()); + for (SSTableReader sstable : sstables) + moved.add(SSTableReader.moveAndOpenSSTable(cfs, sstable.descriptor, cfs.getUniqueDescriptorFor(sstable.descriptor, dst), sstable.getComponents(), true)); + + // 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..c2f871df86 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,17 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized performSSTableImport(AccordExecutor executor, TxnRead.ImportMetadata importMetadata, Timestamp executeAt) + { + Callable callable = () -> { + if (importMetadata.getStreamingEpoch() != executeAt.epoch()) + throw new RuntimeException("SSTable import failed because of a concurrent topology change"); + LocalTransfers.instance.activatePendingTransfers(importMetadata); + 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..6b938f9b50 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java @@ -20,8 +20,10 @@ package org.apache.cassandra.service.accord.txn; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Comparator; import java.util.List; +import java.util.Objects; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -41,6 +43,7 @@ import accord.primitives.Routable.Domain; import accord.primitives.Seekable; import accord.primitives.Seekables; import accord.primitives.Timestamp; +import accord.utils.Invariants; import accord.utils.UnhandledEnum; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; @@ -50,17 +53,21 @@ 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 static accord.primitives.Routables.Slice.Minimal; import static accord.utils.Invariants.require; @@ -87,6 +94,7 @@ public class TxnRead extends AbstractKeySorted implements Read 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; + @Nullable + private final ImportMetadata importMetadata; + private TxnRead(TableMetadatas tables, Domain domain) { super(new TxnNamedRead[0], domain); this.tables = tables; this.domain = domain; this.cassandraConsistencyLevel = null; + this.importMetadata = null; } - 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,7 @@ 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 createSerialRead(List readCommands, ConsistencyLevel consistencyLevel, TableMetadatasAndKeys.KeyCollector keyCollector) @@ -163,13 +177,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 +193,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, TimeUUID[] planIds, long streamingEpoch) + { + ImportMetadata importMetadata = new ImportMetadata(planIds, streamingEpoch); + return new TxnRead(tables, ImmutableList.of(new TxnNamedRead(txnDataName(USER), range, null)), null, importMetadata); } public long estimatedSizeOnHeap() @@ -291,6 +315,11 @@ public class TxnRead extends AbstractKeySorted implements Read throw new UnhandledEnum(select.domain()); } + if (isUsedForImport()) + { + Invariants.require(!reads.isEmpty()); + return new TxnRead(tables, reads, cassandraConsistencyLevel, importMetadata); + } return createTxnRead(tables, reads, cassandraConsistencyLevel, select.domain()); } @@ -372,6 +401,11 @@ public class TxnRead extends AbstractKeySorted implements Read break; } } + if (read.isUsedForImport()) + { + Invariants.require(!reads.isEmpty()); + return new TxnRead(tables, reads, cassandraConsistencyLevel, importMetadata); + } return createTxnRead(tables, reads, cassandraConsistencyLevel, that.domain); } @@ -394,6 +428,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 +444,99 @@ 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 TimeUUID[] planIds; + private final Long streamingEpoch; + + public ImportMetadata(TimeUUID[] planIds, Long streamingEpoch) + { + this.planIds = planIds; + this.streamingEpoch = streamingEpoch; + } + + public TimeUUID[] getPlanIds() + { + return planIds; + } + + public Long getStreamingEpoch() + { + return streamingEpoch; + } + + @Override + public boolean equals(Object that) + { + return that instanceof ImportMetadata && equals((ImportMetadata) that); + } + + public boolean equals(ImportMetadata that) + { + return Arrays.equals(this.planIds, that.planIds) + && (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 + { + serializeArray(importMetadata.getPlanIds(), out, TimeUUID.Serializer.instance); + out.writeLong(importMetadata.streamingEpoch); + } + + public void skip(DataInputPlus in, Version version) throws IOException + { + skipArray(in, TimeUUID.Serializer.instance); + in.readLong(); + } + + @Override + public ImportMetadata deserialize(DataInputPlus in, Version version) throws IOException + { + TimeUUID[] planIds = deserializeArray(in, TimeUUID.Serializer.instance, TimeUUID[]::new); + Long streamingEpoch = in.readLong(); + return new ImportMetadata(planIds, streamingEpoch); + } + + @Override + public long serializedSize(ImportMetadata importMetadata, Version version) + { + long size = 0; + size += serializedArraySize(importMetadata.planIds, TimeUUID.Serializer.instance); + size += TypeSizes.LONG_SIZE; + return size; + } + }; + } + + @Override + public boolean isUsedForImport() + { + return importMetadata != null; + } + 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 +553,18 @@ 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); + case TYPE_IMPORT: + skipArray(tablesAndKeys, in, version, TxnNamedRead.serializer); + deserializeNullable(in, consistencyLevelSerializer); + deserializeNullable(in, version, ImportMetadata.serializer); + default: + throw new IllegalStateException("Unhandled type " + type); } } @@ -453,9 +582,18 @@ 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); + return new TxnRead(tablesAndKeys.tables, items, consistencyLevel, importMetadata); + } } } @@ -463,7 +601,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..84bcf060a5 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java @@ -0,0 +1,484 @@ +/* + * 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.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.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.io.sstable.CQLSSTableWriter; +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.utils.Shared; + +import org.assertj.core.api.Assertions; +import org.junit.Ignore; +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.takesNoArguments; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; +import static org.junit.Assert.assertEquals; + +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 = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + + CQLSSTableWriter.Builder builder1 = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + try (CQLSSTableWriter writer = builder1.build()) + { + writer.addRow(1, 1); + writer.addRow(2, 1); + } + + CQLSSTableWriter.Builder builder2 = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + + try (CQLSSTableWriter writer = builder2.build()) + { + writer.addRow(3, 1); + } + + try (Cluster cluster = init(builder().withNodes(3) + .withoutVNodes() + .withDataDirCount(1) + .withConfig((config) -> + config + .with(Feature.NETWORK, Feature.GOSSIP)).start())) + { + 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(); + }); + }); + + 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 + cluster.forEach(instance -> { + instance.runOnInstance(() -> { + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, "tbl"); + assertEquals(2, cfs.getLiveSSTables().size()); + }); + }); + + // 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 testSSTableImportWithConcurrentTopologyChangeFails() throws Throwable + { + String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + + CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + try (CQLSSTableWriter writer = builder.build()) + { + writer.addRow(1, 1); + writer.addRow(2, 1); + writer.addRow(3, 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())) + { + 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'"); + + 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("SSTable import failed because of a concurrent topology change"); + }); + }, "importer"); + importer.start(); + + cluster.get(1).runOnInstance(() -> { + StorageService.instance.move(Long.toString(Long.parseLong(getOnlyElement(StorageService.instance.getTokens())) + 1)); + State.waitForTopologyChange.countDown(); + }); + + Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS); + + cluster.forEach(instance -> instance.runOnInstance(() -> { + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); + assertEquals(0, cfs.getLiveSSTables().size()); + })); + } + } + + @Test + @Ignore + public void testSSTableImportReplicaDown() throws Throwable + { + String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + + CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + try (CQLSSTableWriter writer = builder.build()) + { + writer.addRow(1, 1); + writer.addRow(2, 1); + writer.addRow(3, 1); + } + + int FAILED_REPLICA = 2; + try (Cluster cluster = init(builder().withNodes(3).withoutVNodes() + .withDataDirCount(1).withConfig((config) -> + config + .with(Feature.NETWORK, Feature.GOSSIP)).start())) + { + 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'"); + + cluster.get(FAILED_REPLICA).shutdown().get(); + + 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)); + }); + + assertPendingDirs(cluster.stream().filter(instance -> instance != cluster.get(FAILED_REPLICA)).collect(Collectors.toList()), (File pendingUuidDir) -> { + Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty(); + }); + + assertLocalSelect(cluster, rows -> assertRows(rows, EMPTY_ROWS)); + } + } + + @Test + public void testSSTableImportStreamingFailedCleanup() throws Throwable + { + String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + + CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + try (CQLSSTableWriter writer = builder.build()) + { + writer.addRow(1, 1); + writer.addRow(2, 1); + writer.addRow(3, 1); + } + + 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())) + { + 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'"); + + 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 + @Ignore + public void testSSTableImportBounceAfterPending() throws Throwable + { + String file = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + + CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + try (CQLSSTableWriter writer = builder.build()) + { + writer.addRow(1, 1); + writer.addRow(2, 1); + writer.addRow(3, 1); + } + + try (Cluster cluster = init(builder().withNodes(3).withoutVNodes() + .withDataDirCount(1).withConfig((config) -> + config + .with(Feature.NETWORK, Feature.GOSSIP)).start())) + { + 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'"); + + // 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 = Files.createTempDirectory(AccordImportSSTableTest.class.getSimpleName()).toString(); + + CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder() + .forTable(TABLE_SCHEMA_CQL) + .inDirectory(file) + .using("INSERT INTO " + KEYSPACE + ".tbl (k, v) " + "VALUES (?, ?)"); + + try (CQLSSTableWriter writer = builder.build()) + { + writer.addRow(1, 1); + writer.addRow(2, 1); + writer.addRow(3, 1); + } + + try (Cluster cluster = init(builder().withNodes(3).withoutVNodes() + .withDataDirCount(1).withConfig((config) -> + config + .with(Feature.NETWORK, Feature.GOSSIP)).start())) + { + 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'"); + + // 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)); + }); + + // Wait until the recovery coordinator picks up the Import Txn + Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS); + + for (int i = 2; i <= 3; i++) + { + cluster.get(i).runOnInstance(() -> { + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE); + assertEquals(1, cfs.getLiveSSTables().size()); + }); + } + + assertLocalSelect(cluster, rows -> { assertRows(rows, row(1, 1), row(2, 1), row(3, 1)); }); + } + } + + 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 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 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 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"); + } + } + } +}