natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

rebase on trunk

This commit is contained in:
Alan Wang 2026-07-17 10:53:19 -07:00
parent 1515da5635
commit c8079bd97f
22 changed files with 1842 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitmodules vendored
View File

@ -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

2
modules/accord

@ -1 +1 @@
Subproject commit fff32de2e915772fbc70d16b1c32346313877838
Subproject commit 53aa05ffcbcf994d25d9239d4a71c493b6c8e98d

27
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -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<SSTableReader> sstables, Consumer<Integer> onRowCacheInvalidation, Consumer<Integer> onCounterCacheInvalidation)
{
if (isRowCacheEnabled() || metadata().isCounter())
{
List<Bounds<Token>> boundsToInvalidate = new ArrayList<>(sstables.size());
sstables.forEach(sstable -> boundsToInvalidate.add(new Bounds<>(sstable.getFirst().getToken(), sstable.getLast().getToken())));
Set<Bounds<Token>> 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<Bounds<Token>> boundsToInvalidate)
{
int invalidatedKeys = 0;

41
src/java/org/apache/cassandra/db/Directories.java
View File

@ -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<File> getPendingLocations()
{
Set<File> 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();

14
src/java/org/apache/cassandra/db/SSTableImporter.java
View File

@ -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<Pair<Directories.SSTableLister, String>> listers = getSSTableListers(options.srcPaths);
Set<Descriptor> 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())

4
src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
View File

@ -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)

12
src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java
View File

@ -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);
}

13
src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
View File

@ -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

44
src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java Normal file
View File

@ -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);
}
}

15
src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java
View File

@ -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<PartitionPosition> boundaries;
private final List<Directories.DataDirectory> directories;
protected final List<Directories.DataDirectory> 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<SSTableMultiWriter> 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;
}

5
src/java/org/apache/cassandra/net/Verb.java
View File

@ -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 ),

25
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -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<SSTableReader> sstables, TableMetadata metadata)
{
logger.info("Creating Accord bulk transfer for keyspace '{}' table '{}' SSTables {}...", keyspace, metadata.name, sstables);
Topology topology = topology().current();
Map<InetAddressAndPort, CoordinatedTransfer.SSTablesForNode> 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);
}
}

499
src/java/org/apache/cassandra/service/accord/CoordinatedTransfer.java Normal file
View File

@ -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.
* <p>
* The transfer proceeds through the following two phases:
* <ol>
* <li>
* <b>Streaming</b>
* 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.
* </li>
* <li>
* <b>ImportTxn</b>
* 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.
* </li>
* </ol>
*
* 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.
* <p>
* 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<InetAddressAndPort, SSTablesForNode> nodeStreamingContext;
final ConcurrentMap<InetAddressAndPort, SingleTransferResult> streamResults;
public CoordinatedTransfer(Long id, String keyspace, TableMetadata tableMetadata, Map<InetAddressAndPort, SSTablesForNode> 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<InetAddressAndPort, SingleTransferResult> 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<TxnResult> accordResult = AccordService.instance().coordinateAsync(tableMetadata.epoch.getEpoch(), txn, ConsistencyLevel.ALL, Dispatcher.RequestTime.forImmediateExecution());
accordResult.awaitAndGet();
}
private void stream()
{
List<Future<Void>> streaming = new ArrayList<>(streamResults.size());
for (InetAddressAndPort to : streamResults.keySet())
{
Future<Void> 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<Throwable> failures = null;
for (Future<Void> 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<Range<Token>> ranges = nodeStreamingContext.get(to).ranges;
for (Map.Entry<SSTableReader, List<SSTableReader.PartitionPositionBounds>> entry : sstablesForNode.positionsForSSTables.entrySet())
{
SSTableReader sstable = entry.getKey();
List<SSTableReader.PartitionPositionBounds> 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<Void> implements RequestCallbackWithFailure<NoPayload>
{
final Set<InetAddressAndPort> responses = ConcurrentHashMap.newKeySet(streamResults.size());
@Override
public void onResponse(Message<NoPayload> 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<InetAddressAndPort, Message<TransferFailed>> msgs = new HashMap<>();
NotifyFailure notifyFailure = new NotifyFailure();
for (Map.Entry<InetAddressAndPort, SingleTransferResult> 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<InetAddressAndPort,Message<TransferFailed>> entry : msgs.entrySet())
MessagingService.instance().sendWithCallback(entry.getValue(), entry.getKey(), notifyFailure);
notifyFailure.get();
}
public static class SSTablesForNode
{
final Node.Id id;
final List<Range<Token>> ranges;
final Map<SSTableReader, List<SSTableReader.PartitionPositionBounds>> positionsForSSTables;
public SSTablesForNode(Node.Id id, Map<SSTableReader, List<SSTableReader.PartitionPositionBounds>> positionsForSSTables, List<Range<Token>> 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<SSTableReader> 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<InetAddressAndPort, SSTablesForNode> getNodeStreamingContext(Collection<SSTableReader> sstables, Topology topology, AccordEndpointMapper endpointMapper)
{
Map<InetAddressAndPort, SSTablesForNode> nodeStreamingContext = new HashMap<>();
for (Id nodeId : topology.nodes())
{
Ranges rangesForNode = topology.rangesForNode(nodeId);
List<Range<Token>> ranges = new ArrayList<>();
for (accord.primitives.Range range : rangesForNode)
ranges.add(((TokenRange) range).toKeyspaceRange());
Map<SSTableReader, List<SSTableReader.PartitionPositionBounds>> positionsForSSTables = new HashMap<>();
for (SSTableReader sstable : sstables)
{
List<SSTableReader.PartitionPositionBounds> 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;
}
}

31
src/java/org/apache/cassandra/service/accord/IAccordService.java
View File

@ -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<SSTableReader> 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<SSTableReader> 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<SSTableReader> sstables, TableMetadata metadata)
{
delegate.executeTransfer(keyspace, sstables, metadata);
}
@Override
public void receivedSSTableImport(PendingLocalTransfer transfer)
{
delegate.receivedSSTableImport(transfer);
}
@Override
public void ensureMinHlc(long minHlc)
{

253
src/java/org/apache/cassandra/service/accord/LocalTransfers.java Normal file
View File

@ -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<Long, CoordinatedTransfer> coordinating = new ConcurrentHashMap<>();
// Added when we have a streamed SSTable in our pending directory
private final Map<TimeUUID, PendingLocalTransfer> 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/<planId>/
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<TransferFailed> verbHandler = message -> {
LocalTransfers.instance().purge(message.payload);
MessagingService.instance().respond(NoPayload.noPayload, message);
};
}

132
src/java/org/apache/cassandra/service/accord/PendingLocalTransfer.java Normal file
View File

@ -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<SSTableReader> sstables;
final long createdAt = currentTimeMillis();
transient String keyspace;
volatile boolean activated = false;
public PendingLocalTransfer(TableId tableId, TimeUUID planId, Collection<SSTableReader> 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<SSTableReader> 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<Integer> onRowCacheInvalidation = invalidatedKeys -> {
logger.debug("{} Invalidated {} row cache entries on table {}.{} after activating transfer",
logPrefix(), invalidatedKeys, cfs.getKeyspaceName(), cfs.getTableName());
};
Consumer<Integer> 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);
}
}

72
src/java/org/apache/cassandra/service/accord/TransferFailed.java Normal file
View File

@ -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<TransferFailed> serializer = new IVersionedSerializer<TransferFailed>()
{
@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 +
'}';
}
}

19
src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
View File

@ -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<ReadC
this.key = key;
}
TxnNamedRead(int name, TokenRange range, ByteBuffer bytes)
{
super(bytes);
this.name = name;
this.key = range;
}
public static TokenRange boundsAsAccordRange(AbstractBounds<PartitionPosition> range, TableId tableId)
{
// Should already have been unwrapped
@ -402,6 +410,17 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized<ReadC
return submit(executor, callable, callable);
}
public AsyncChain<Data> performSSTableImport(AccordExecutor executor, TxnRead.ImportMetadata importMetadata, Timestamp executeAt)
{
Callable<Data> 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<Data> submit(AccordExecutor executor, Callable<Data> readCallable, Object describe)
{
return executor.buildDebuggable(readCallable, describe);

168
src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
View File

@ -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<TxnNamedRead> 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<TxnNamedRead> 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<TxnNamedRead> items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
private TxnRead(TableMetadatas tables, @Nonnull List<TxnNamedRead> 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<TxnNamedRead> 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<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel, TableMetadatasAndKeys.KeyCollector keyCollector)
@ -163,13 +177,13 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> 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<TxnNamedRead> implements Read
List<TxnNamedRead> 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<PartitionPosition> 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<TxnNamedRead> 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<TxnNamedRead> 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<TxnNamedRead> 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<TxnNamedRead> 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<ImportMetadata, Version> 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<TxnRead, TableMetadatasAndKeys, Version> 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<TxnNamedRead> 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<TxnNamedRead> 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<TxnNamedRead> 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);

3
src/java/org/apache/cassandra/streaming/StreamOperation.java
View File

@ -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;

4
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -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;

484
test/distributed/org/apache/cassandra/distributed/test/accord/AccordImportSSTableTest.java Normal file
View File

@ -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<String> 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<String> 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<String> 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<String> 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<String> 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<String> 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<IInvokableInstance> validate, IIsolatedExecutor.SerializableConsumer<File> forPendingUuidDir)
{
for (IInvokableInstance instance : validate)
{
instance.runOnInstance(() -> {
Set<File> 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<IInvokableInstance> validate, IIsolatedExecutor.SerializableConsumer<Object[][]> 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<Void> 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");
}
}
}
}