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

CEP-45: Bulk transfer 

patch by Abe Ratnofsky; reviewed by Caleb Rackliffe and Blake Eggleston for CASSANDRA-20383
This commit is contained in:
Abe Ratnofsky 2025-09-02 13:07:38 -04:00 committed by Blake Eggleston
parent 129cbf3f75
commit 407e19021b
111 changed files with 5013 additions and 230 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -950,7 +950,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
@ -1739,6 +1739,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
CompactionManager.instance.submitBackground(this);
}
/**
* addSStables variant that allows adding sstables for tracked tables without requiring tracked transfers during bootstrap
*/
public void addSSTableForBootstrap(Collection<SSTableReader> sstables)
{
data.addSSTablesForBootstrap(sstables);
logger.debug("Adding sstables {}", sstables);
CompactionManager.instance.submitBackground(this);
}
/**
* Calculate expected file size of SSTable after compaction.
*
@ -2198,6 +2208,30 @@ 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;

43
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;
@ -116,6 +117,7 @@ public class Directories
public static final String BACKUPS_SUBDIR = "backups";
public static final String SNAPSHOT_SUBDIR = "snapshots";
public static final String PENDING_SUBDIR = "pending";
public static final String TMP_SUBDIR = "tmp";
public static final String SECONDARY_INDEX_NAME_SEPARATOR = ".";
public static final String TABLE_DIRECTORY_NAME_SEPARATOR = "-";
@ -315,13 +317,8 @@ 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 +723,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 +838,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();

2
src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
View File

@ -378,6 +378,8 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
public UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController controller)
{
ColumnFamilyStore.ViewFragment view = cfs.select(View.selectLive(dataRange().keyRange()));
if (cfs.metadata().replicationType().isTracked())
controller.addTransferIds(view.sstables);
Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().partitionKeyType));
// fetch data from current memtable, historical memtables, and SSTables in the correct order.

48
src/java/org/apache/cassandra/db/ReadExecutionController.java
View File

@ -18,21 +18,37 @@
package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore.ViewFragment;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.replication.ActivatedTransfers;
import org.apache.cassandra.replication.ShortMutationId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.MonotonicClock;
import org.apache.cassandra.utils.concurrent.OpOrder;
import static org.apache.cassandra.utils.MonotonicClock.Global.preciseTime;
@NotThreadSafe
public class ReadExecutionController implements AutoCloseable
{
private static final Logger logger = LoggerFactory.getLogger(ReadExecutionController.class);
private static final long NO_SAMPLING = Long.MIN_VALUE;
// For every reads
@ -50,6 +66,13 @@ public class ReadExecutionController implements AutoCloseable
private final RepairedDataInfo repairedDataInfo;
private long oldestUnrepairedTombstone = Long.MAX_VALUE;
/**
* Track bulk transfers involved in the read, so we can do read reconciliation. These come from the
* {@link ViewFragment}, not the SSTable read path, so bloom filters and short-circuiting SSTable scans will still
* include the total set of relevant bulk transfers.
*/
private Set<ShortMutationId> transferIds = null;
ReadExecutionController(ReadCommand command,
OpOrder.Group baseOp,
TableMetadata baseMetadata,
@ -243,4 +266,29 @@ public class ReadExecutionController implements AutoCloseable
if (cfs != null)
cfs.metric.topLocalReadQueryTime.addSample(cql, timeMicros);
}
public void addTransferIds(Iterable<SSTableReader> sstables)
{
Preconditions.checkState(metadata().replicationType().isTracked());
if (transferIds == null)
transferIds = new HashSet<>();
for (SSTableReader sstable : sstables)
{
ActivatedTransfers transfers = sstable.getCoordinatorLogOffsets().transfers();
if (transfers.isEmpty())
continue;
logger.trace("Adding overlapping IDs to read keyRange {}", command.dataRange.keyRange);
transfers.forEachIntersecting(command.dataRange.keyRange, id -> {
String msg = String.format("Found overlapping activation ID %s for queried range %s", id, command.dataRange.keyRange);
Tracing.trace(msg);
logger.debug(msg);
transferIds.add(id);
});
}
}
public Iterator<ShortMutationId> getTransferIds()
{
return transferIds == null ? null : transferIds.iterator();
}
}

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

@ -45,6 +45,7 @@ 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.replication.MutationTrackingService;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageService;
@ -81,11 +82,8 @@ public class SSTableImporter
UUID importID = UUID.randomUUID();
logger.info("[{}] Loading new SSTables for {}/{}: {}", importID, cfs.getKeyspaceName(), cfs.getTableName(), options);
// This will be supported in the future
TableMetadata metadata = cfs.metadata();
if (metadata.replicationType() != null && metadata.replicationType().isTracked())
throw new IllegalStateException("Can't import into tables with mutation tracking enabled");
boolean isTracked = metadata.replicationType().isTracked();
List<Pair<Directories.SSTableLister, String>> listers = getSSTableListers(options.srcPaths);
Set<Descriptor> currentDescriptors = new HashSet<>();
@ -184,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 = isTracked
? SSTableReader.open(cfs, oldDescriptor, metadata.ref)
: SSTableReader.moveAndOpenSSTable(cfs, oldDescriptor, newDescriptor, entry.getValue(), options.copyData);
newSSTablesPerDirectory.add(sstable);
}
catch (Throwable t)
@ -234,12 +235,14 @@ public class SSTableImporter
if (!cfs.indexManager.validateSSTableAttachedIndexes(newSSTables, false, options.validateIndexChecksum))
cfs.indexManager.buildSSTableAttachedIndexesBlocking(newSSTables);
cfs.getTracker().addSSTables(newSSTables);
if (isTracked)
TrackedBulkTransfer.execute(cfs.keyspace.getName(), newSSTables);
else
cfs.getTracker().addSSTables(newSSTables);
for (SSTableReader reader : newSSTables)
{
if (options.invalidateCaches && cfs.isRowCacheEnabled())
invalidateCachesForSSTable(reader);
}
}
catch (Throwable t)
{
@ -251,6 +254,17 @@ public class SSTableImporter
return failedDirectories;
}
/**
* TODO: Support user-defined consistency level for import, for import with replicas down
*/
private static class TrackedBulkTransfer
{
private static void execute(String keyspace, Set<SSTableReader> sstables)
{
MutationTrackingService.instance.executeTransfers(keyspace, sstables, ConsistencyLevel.ALL);
}
}
/**
* Check the state of this node and throws an {@link InterruptedException} if it is currently draining
*

7
src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
View File

@ -783,6 +783,8 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
Tracing.trace("Acquiring sstable references");
List<SSTableReader> sstables = view.sstables();
sstables.sort(SSTableReader.maxTimestampDescending);
if (cfs.metadata().replicationType().isTracked())
controller.addTransferIds(sstables);
ClusteringIndexFilter filter = clusteringIndexFilter();
long minTimestamp = Long.MAX_VALUE;
long mostRecentPartitionTombstone = Long.MIN_VALUE;
@ -1063,6 +1065,11 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
/* add the SSTables on disk */
List<SSTableReader> sstables = view.sstables();
sstables.sort(SSTableReader.maxTimestampDescending);
if (cfs.metadata().replicationType().isTracked())
{
logger.trace("Executing read against SSTables {}", sstables);
controller.addTransferIds(view.sstables());
}
// read sorted sstables
for (SSTableReader sstable : sstables)
{

2
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -2042,7 +2042,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan
fullWriter.commit();
unrepairedWriter.commit();
txn.commit();
logger.info("Anticompacted {} in {}.{} to full = {}, transient = {}, unrepaired = {} for {}",
logger.info("Anticompacted {} in {}.{} to full = {}, unrepaired = {} for {}",
sstableAsSet,
cfs.getKeyspaceName(),
cfs.getTableName(),

2
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -56,6 +56,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.snapshot.SnapshotManager;
import org.apache.cassandra.service.snapshot.SnapshotOptions;
@ -447,6 +448,7 @@ public class CompactionTask extends AbstractCompactionTask
ImmutableCoordinatorLogOffsets.Builder builder = new ImmutableCoordinatorLogOffsets.Builder();
for (SSTableReader sstable : sstables)
builder.addAll(sstable.getCoordinatorLogOffsets());
builder.purgeTransfers(MutationTrackingService.instance::isDurablyReconciled);
return builder.build();
}

24
src/java/org/apache/cassandra/db/lifecycle/Tracker.java
View File

@ -27,6 +27,7 @@ import java.util.concurrent.locks.ReentrantLock;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.Iterables;
@ -62,6 +63,7 @@ import org.apache.cassandra.notifications.SSTableRepairStatusChanged;
import org.apache.cassandra.notifications.TableDroppedNotification;
import org.apache.cassandra.notifications.TablePreScrubNotification;
import org.apache.cassandra.notifications.TruncationNotification;
import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.TimeUUID;
@ -270,6 +272,28 @@ public class Tracker
}
public void addSSTables(Collection<SSTableReader> sstables)
{
Preconditions.checkState(!cfstore.metadata().replicationType().isTracked());
addSSTablesInternal(sstables, false, true, true);
}
public void addSSTablesTracked(Collection<SSTableReader> sstables)
{
Preconditions.checkState(cfstore.metadata().replicationType().isTracked());
for (SSTableReader sstable : sstables)
{
ImmutableCoordinatorLogOffsets logOffsets = sstable.getCoordinatorLogOffsets();
Preconditions.checkState(logOffsets.mutations().isEmpty());
Preconditions.checkState(!logOffsets.transfers().isEmpty());
}
addSSTablesInternal(sstables, false, true, true);
}
/**
* addSStables variant for bootstrap. Doesn't validate tracked/untracked or absence of logOffsets data
*/
public void addSSTablesForBootstrap(Collection<SSTableReader> sstables)
{
addSSTablesInternal(sstables, false, true, true);
}

2
src/java/org/apache/cassandra/db/lifecycle/View.java
View File

@ -403,4 +403,4 @@ public class View
}
};
}
}
}

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

@ -48,6 +48,7 @@ import org.apache.cassandra.streaming.ProgressInfo;
import org.apache.cassandra.streaming.StreamReceiver;
import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.streaming.messages.StreamMessageHeader;
import org.apache.cassandra.utils.FBUtilities;
import static java.lang.String.format;
import static org.apache.cassandra.utils.FBUtilities.prettyPrintMemory;
@ -139,7 +140,7 @@ public class CassandraEntireSSTableStreamReader implements IStreamReader
UnaryOperator<StatsMetadata> transform = stats -> stats.mutateLevel(header.sstableLevel)
.mutateRepairedMetadata(messageHeader.repairedAt, messageHeader.pendingRepair);
String description = String.format("level %s and repairedAt time %s and pendingRepair %s",
String description = format("level %s and repairedAt time %s and pendingRepair %s",
header.sstableLevel, messageHeader.repairedAt, messageHeader.pendingRepair);
writer.descriptor().getMetadataSerializer().mutate(writer.descriptor(), description, transform);
return writer;
@ -159,9 +160,13 @@ public class CassandraEntireSSTableStreamReader implements IStreamReader
private File getDataDir(ColumnFamilyStore cfs, long totalSize) throws IOException
{
boolean isTracked = cfs.metadata().replicationType().isTracked();
Directories.DataDirectory localDir = cfs.getDirectories().getWriteableLocation(totalSize);
if (localDir == null)
throw new IOException(format("Insufficient disk space to store %s", prettyPrintMemory(totalSize)));
throw new IOException(format("Insufficient disk space to store %s", FBUtilities.prettyPrintMemory(totalSize)));
if (isTracked)
return cfs.getDirectories().getPendingLocationForDisk(localDir, session.planId());
File dir = cfs.getDirectories().getLocationForDisk(cfs.getDiskBoundaries().getCorrectDiskForKey(header.firstKey));

13
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;
@ -62,6 +63,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;
@ -188,7 +190,16 @@ 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, coordinatorLogOffsets, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
RangeAwareSSTableWriter writer;
if (session.streamOperation() == StreamOperation.TRACKED_TRANSFER)
{
Preconditions.checkState(cfs.metadata().replicationType().isTracked());
writer = new RangeAwarePendingSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, coordinatorLogOffsets, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()), session.planId());
}
else
writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, pendingRepair, coordinatorLogOffsets, format, sstableLevel, totalSize, txn, getHeader(cfs.metadata()));
return new SSTableTxnSingleStreamWriter(txn, writer);
}

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

@ -21,12 +21,13 @@ package org.apache.cassandra.db.streaming;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.function.Consumer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import org.apache.cassandra.streaming.StreamOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -42,7 +43,6 @@ import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.ThrottledUnfilteredIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.view.View;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.ISSTableScanner;
@ -53,7 +53,10 @@ import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.AccordTopology;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.TimeOnlyRequestBookkeeping.LatencyRequestBookkeeping;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.replication.PendingLocalTransfer;
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;
@ -133,6 +136,13 @@ public class CassandraStreamReceiver implements StreamReceiver
txn.update(finished);
sstables.addAll(finished);
receivedEntireSSTable = file.isEntireSSTable();
if (session.streamOperation() == StreamOperation.TRACKED_TRANSFER)
{
Preconditions.checkState(cfs.metadata().replicationType().isTracked());
PendingLocalTransfer transfer = new PendingLocalTransfer(cfs.metadata().id, session.planId(), sstables);
MutationTrackingService.instance.received(transfer);
}
}
@Override
@ -257,33 +267,31 @@ 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);
cfs.addSSTables(readers);
//invalidate row and counter cache
if (cfs.isRowCacheEnabled() || cfs.metadata().isCounter())
// Don't mark as live until activated by the stream coordinator
if (session.streamOperation() == StreamOperation.TRACKED_TRANSFER)
return;
if (session.streamOperation() == StreamOperation.BOOTSTRAP)
{
List<Bounds<Token>> boundsToInvalidate = new ArrayList<>(readers.size());
readers.forEach(sstable -> boundsToInvalidate.add(new Bounds<Token>(sstable.getFirst().getToken(), sstable.getLast().getToken())));
Set<Bounds<Token>> nonOverlappingBounds = Bounds.getNonOverlappingBounds(boundsToInvalidate);
if (cfs.isRowCacheEnabled())
{
int invalidatedKeys = cfs.invalidateRowCache(nonOverlappingBounds);
if (invalidatedKeys > 0)
logger.debug("[Stream #{}] Invalidated {} row cache entries on table {}.{} after stream " +
"receive task completed.", session.planId(), invalidatedKeys,
cfs.getKeyspaceName(), cfs.getTableName());
}
if (cfs.metadata().isCounter())
{
int invalidatedKeys = cfs.invalidateCounterCache(nonOverlappingBounds);
if (invalidatedKeys > 0)
logger.debug("[Stream #{}] Invalidated {} counter cache entries on table {}.{} after stream " +
"receive task completed.", session.planId(), invalidatedKeys,
cfs.getKeyspaceName(), cfs.getTableName());
}
cfs.addSSTableForBootstrap(readers);
}
else
{
cfs.addSSTables(readers);
}
Consumer<Integer> onRowCacheInvalidation = invalidatedKeys -> {
logger.debug("[Stream #{}] Invalidated {} row cache entries on table {}.{} after stream " +
"receive task completed.", session.planId(), invalidatedKeys,
cfs.getKeyspaceName(), cfs.getTableName());
};
Consumer<Integer> onCounterCacheInvalidation = invalidatedKeys -> {
logger.debug("[Stream #{}] Invalidated {} counter cache entries on table {}.{} after stream " +
"receive task completed.", session.planId(), invalidatedKeys,
cfs.getKeyspaceName(), cfs.getTableName());
};
cfs.invalidateRowAndCounterCache(readers, onRowCacheInvalidation, onCounterCacheInvalidation);
}
}
}

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

@ -0,0 +1,45 @@
/*
* 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.replication.ImmutableCoordinatorLogOffsets;
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, ImmutableCoordinatorLogOffsets coordinatorLogOffsets, SSTableFormat<?, ?> format, int sstableLevel, long totalSize, ILifecycleTransaction txn, SerializationHeader header, TimeUUID planId) throws IOException
{
super(cfs, estimatedKeys, repairedAt, pendingRepair, coordinatorLogOffsets, format, sstableLevel, totalSize, txn, header);
this.planId = planId;
}
@Override
protected File currentFile()
{
return cfs.getDirectories().getPendingLocationForDisk(directories.get(currentIndex), planId);
}
}

14
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.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.FBUtilities;
@ -40,7 +41,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;
@ -49,7 +50,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;
@ -79,7 +80,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
}
}
private void maybeSwitchWriter(DecoratedKey key)
protected void maybeSwitchWriter(DecoratedKey key)
{
if (boundaries == null)
return;
@ -96,11 +97,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, coordinatorLogOffsets, null, sstableLevel, header, txn);
}
}
protected File currentFile()
{
return cfs.getDirectories().getLocationForDisk(directories.get(currentIndex));
}
public void append(UnfilteredRowIterator partition)
{
maybeSwitchWriter(partition.partitionKey());

12
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -1401,6 +1401,18 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource,
}
}
/**
* Mutate {@link ImmutableCoordinatorLogOffsets} with a lock to avoid racing with entire-sstable-streaming and then reload sstable metadata
*/
public void mutateCoordinatorLogOffsetsAndReload(ImmutableCoordinatorLogOffsets logOffsets) throws IOException
{
synchronized (tidy.global)
{
descriptor.getMetadataSerializer().mutateCoordinatorLogOffsets(descriptor, logOffsets);
reloadSSTableMetadata();
}
}
/**
* Reloads the sstable metadata from disk.
* <p>

6
src/java/org/apache/cassandra/io/sstable/metadata/IMetadataSerializer.java
View File

@ -26,6 +26,7 @@ import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.utils.TimeUUID;
/**
@ -96,6 +97,11 @@ public interface IMetadataSerializer
*/
public void mutateRepairMetadata(Descriptor descriptor, long newRepairedAt, TimeUUID newPendingRepair) throws IOException;
/**
* Replace mutation tracking metadata.
*/
public void mutateCoordinatorLogOffsets(Descriptor descriptor, ImmutableCoordinatorLogOffsets logOffsets) throws IOException;
/**
* Replace the sstable metadata file ({@code -Statistics.db}) with the given components.
*/

10
src/java/org/apache/cassandra/io/sstable/metadata/MetadataSerializer.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.io.util.DataOutputStreamPlus;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
@ -249,6 +250,15 @@ public class MetadataSerializer implements IMetadataSerializer
mutate(descriptor, stats -> stats.mutateRepairedMetadata(newRepairedAt, newPendingRepair));
}
@Override
public void mutateCoordinatorLogOffsets(Descriptor descriptor, ImmutableCoordinatorLogOffsets logOffsets) throws IOException
{
if (logger.isTraceEnabled())
logger.trace("Mutating {} to {}", descriptor.fileFor(Components.STATS), logOffsets);
mutate(descriptor, stats -> stats.mutateCoordinatorLogOffsets(logOffsets));
}
private void mutate(Descriptor descriptor, UnaryOperator<StatsMetadata> transform) throws IOException
{
Map<MetadataType, MetadataComponent> currentComponents = deserialize(descriptor, EnumSet.allOf(MetadataType.class));

29
src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
View File

@ -241,6 +241,35 @@ public class StatsMetadata extends MetadataComponent
lastKey);
}
public StatsMetadata mutateCoordinatorLogOffsets(ImmutableCoordinatorLogOffsets newLogOffsets)
{
return new StatsMetadata(estimatedPartitionSize,
estimatedCellPerPartitionCount,
commitLogIntervals,
minTimestamp,
maxTimestamp,
minLocalDeletionTime,
maxLocalDeletionTime,
minTTL,
maxTTL,
compressionRatio,
estimatedTombstoneDropTime,
sstableLevel,
clusteringTypes,
coveredClustering,
hasLegacyCounterShards,
repairedAt,
totalColumnsSet,
totalRows,
tokenSpaceCoverage,
originatingHostId,
pendingRepair,
hasPartitionLevelDeletions,
newLogOffsets,
firstKey,
lastKey);
}
@Override
public boolean equals(Object o)
{

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

@ -79,8 +79,11 @@ import org.apache.cassandra.repair.messages.ValidationRequest;
import org.apache.cassandra.repair.messages.ValidationResponse;
import org.apache.cassandra.replication.BroadcastLogOffsets;
import org.apache.cassandra.replication.ForwardedWrite;
import org.apache.cassandra.replication.LocalTransfers;
import org.apache.cassandra.replication.PullMutationsRequest;
import org.apache.cassandra.replication.PushMutationRequest;
import org.apache.cassandra.replication.TransferActivation;
import org.apache.cassandra.replication.TransferFailed;
import org.apache.cassandra.schema.SchemaMutationsSerializer;
import org.apache.cassandra.schema.SchemaPullVerbHandler;
import org.apache.cassandra.schema.SchemaPushVerbHandler;
@ -341,6 +344,10 @@ public enum Verb
TRACKED_SUMMARY_RSP (910, P2, readTimeout, REQUEST_RESPONSE, () -> TrackedSummaryResponse.serializer, () -> TrackedSummaryResponse.verbHandler ),
TRACKED_SUMMARY_REQ (911, P3, readTimeout, READ, () -> TrackedRead.SummaryRequest.serializer, () -> TrackedRead.verbHandler, TRACKED_SUMMARY_RSP ),
TRACKED_TRANSFER_ACTIVATE_RSP (912, P1, repairTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER),
TRACKED_TRANSFER_ACTIVATE_REQ (913, P1, repairTimeout, ANTI_ENTROPY, () -> TransferActivation.serializer, () -> TransferActivation.verbHandler, TRACKED_TRANSFER_ACTIVATE_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),
// accord
ACCORD_SIMPLE_RSP (119, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(EnumSerializer.simpleReply), AccordService::responseHandlerOrNoop ),

287
src/java/org/apache/cassandra/replication/ActivatedTransfers.java Normal file
View File

@ -0,0 +1,287 @@
/*
* 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.replication;
import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Comparators;
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.CollectionSerializers;
import org.apache.cassandra.utils.IntervalTree;
/**
* A collection of activated bulk transfers. Bulk transfers are accessed for overlapping {@link ShortMutationId} for a
* given {@link Bounds}, to find the set of transfer IDs that intersect with a given read, for example. Tracked reads
* must include intersecting transfers to ensure that summaries reflect transferred SSTables.
* <p>
* We expect to have very few bulk transfers (typically 0) so they're kept in an un-indexed set. If we have more
* transfers in the future, we could transition this to an {@link IntervalTree}.
*/
public class ActivatedTransfers implements Iterable<ShortMutationId>
{
public static final ActivatedTransfers EMPTY = new ActivatedTransfers();
private final Set<ActivatedTransfer> transfers;
public ActivatedTransfers()
{
this(new HashSet<>(1));
}
private ActivatedTransfers(Collection<ActivatedTransfer> transfers)
{
this.transfers = new HashSet<>(transfers);
}
public static ActivatedTransfers copyOf(ActivatedTransfers other)
{
return other == null ? new ActivatedTransfers() : new ActivatedTransfers(other.transfers);
}
@VisibleForTesting
static final class ActivatedTransfer
{
final ShortMutationId id;
final Bounds<Token> bounds;
@VisibleForTesting
ActivatedTransfer(ShortMutationId id, Bounds<Token> bounds)
{
this.id = id;
this.bounds = bounds;
}
private ActivatedTransfer(ShortMutationId id, Collection<SSTableReader> sstables)
{
this(id, covering(sstables));
}
public static final IVersionedSerializer<ActivatedTransfer> serializer = new IVersionedSerializer<>()
{
@Override
public void serialize(ActivatedTransfer transfer, DataOutputPlus out, int version) throws IOException
{
ShortMutationId.serializer.serialize(transfer.id, out, version);
Token.serializer.serialize(transfer.bounds.left, out, version);
Token.serializer.serialize(transfer.bounds.right, out, version);
}
@Override
public ActivatedTransfer deserialize(DataInputPlus in, int version) throws IOException
{
ShortMutationId id = ShortMutationId.serializer.deserialize(in, version);
Token left = Token.serializer.deserialize(in, version);
Token right = Token.serializer.deserialize(in, version);
return new ActivatedTransfer(id, new Bounds<Token>(left, right));
}
@Override
public long serializedSize(ActivatedTransfer transfer, int version)
{
long size = 0;
size += ShortMutationId.serializer.serializedSize(transfer.id, version);
size += Token.serializer.serializedSize(transfer.bounds.left, version);
size += Token.serializer.serializedSize(transfer.bounds.right, version);
return size;
}
};
@Override
public String toString()
{
return "ActivatedTransfer{" +
"id=" + id +
", bounds=" + bounds +
'}';
}
@Override
public boolean equals(Object o)
{
if (o == null || getClass() != o.getClass()) return false;
ActivatedTransfer that = (ActivatedTransfer) o;
return Objects.equals(id, that.id) && Objects.equals(bounds, that.bounds);
}
@Override
public int hashCode()
{
return Objects.hash(id, bounds);
}
}
public void removeOffset(int offset)
{
transfers.removeIf(transfer -> transfer.id.offset() == offset);
}
@VisibleForTesting
public void add(ShortMutationId transferId, Bounds<Token> bounds)
{
transfers.add(new ActivatedTransfer(transferId, bounds));
}
public void add(ShortMutationId transferId, Collection<SSTableReader> sstables)
{
transfers.add(new ActivatedTransfer(transferId, sstables));
}
public void addAll(ActivatedTransfers other)
{
transfers.addAll(other.transfers);
}
public void forEachIntersecting(AbstractBounds<PartitionPosition> range, Consumer<ShortMutationId> consumer)
{
for (ActivatedTransfer transfer : transfers)
if (intersects(transfer.bounds, range))
consumer.accept(transfer.id);
}
public void forEachIntersecting(Token token, Consumer<ShortMutationId> consumer)
{
for (ActivatedTransfer transfer : transfers)
if (transfer.bounds.contains(token))
consumer.accept(transfer.id);
}
@Override
public Iterator<ShortMutationId> iterator()
{
return Iterators.transform(transfers.iterator(), transfer -> transfer.id);
}
public boolean isEmpty()
{
return transfers.isEmpty();
}
private static Bounds<Token> covering(Collection<SSTableReader> sstables)
{
Preconditions.checkArgument(!sstables.isEmpty());
Iterator<SSTableReader> iter = sstables.iterator();
SSTableReader next = iter.next();
Token left = next.getFirst().getToken();
Token right = next.getLast().getToken();
while (iter.hasNext())
{
next = iter.next();
left = Comparators.min(left, next.getFirst().getToken());
right = Comparators.max(right, next.getLast().getToken());
}
return new Bounds<>(left, right);
}
private static boolean intersects(Bounds<Token> bounds, AbstractBounds<PartitionPosition> range)
{
Preconditions.checkArgument(!AbstractBounds.strictlyWrapsAround(bounds.left, bounds.right));
if (range instanceof Range && ((Range<?>) range).isTrulyWrapAround())
{
List<? extends AbstractBounds<PartitionPosition>> unwrapped = range.unwrap();
return Iterables.any(unwrapped, unwrap -> intersects(bounds, unwrap));
}
if (range.right.getToken().isMinimum())
{
/*
bounds: []
range: ?----|
*/
boolean overlapsPastBoundary = bounds.right.compareTo(range.left.getToken()) > 0;
/*
bounds: []
range: [----|
*/
boolean overlapsAtBoundary = bounds.right.equals(range.left.getToken()) && range.inclusiveLeft();
return overlapsPastBoundary || overlapsAtBoundary;
}
if ((range.left.getToken().compareTo(bounds.right) < 0) && (bounds.left.compareTo(range.right.getToken()) < 0))
return true;
if (range.inclusiveLeft() && bounds.contains(range.left.getToken()))
return true;
if (range.inclusiveRight() && bounds.contains(range.right.getToken()))
return true;
return false;
}
@Override
public boolean equals(Object o)
{
if (o == null || getClass() != o.getClass()) return false;
ActivatedTransfers that = (ActivatedTransfers) o;
return Objects.equals(transfers, that.transfers);
}
@Override
public int hashCode()
{
return Objects.hashCode(transfers);
}
@Override
public String toString()
{
return "ActivatedTransfers{" +
"transfers=" + transfers +
'}';
}
public static final IVersionedSerializer<ActivatedTransfers> serializer = new IVersionedSerializer<>()
{
@Override
public void serialize(ActivatedTransfers transfers, DataOutputPlus out, int version) throws IOException
{
CollectionSerializers.serializeCollection(transfers.transfers, out, version, ActivatedTransfer.serializer);
}
@Override
public ActivatedTransfers deserialize(DataInputPlus in, int version) throws IOException
{
return new ActivatedTransfers(CollectionSerializers.deserializeSet(in, version, ActivatedTransfer.serializer));
}
@Override
public long serializedSize(ActivatedTransfers transfers, int version)
{
return CollectionSerializers.serializedCollectionSize(transfers.transfers, version, ActivatedTransfer.serializer);
}
};
}

81
src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
View File

@ -17,11 +17,15 @@
*/
package org.apache.cassandra.replication;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.agrona.concurrent.ManyToOneConcurrentLinkedQueue;
import org.apache.cassandra.concurrent.Interruptible;
import org.apache.cassandra.concurrent.Shutdownable;
@ -46,6 +50,8 @@ import static org.apache.cassandra.utils.concurrent.Semaphore.newSemaphore;
// TODO (expected): handle temporarily down nodes
public final class ActiveLogReconciler implements Shutdownable
{
private static final Logger logger = LoggerFactory.getLogger(ActiveLogReconciler.class);
public enum Priority { HIGH, REGULAR }
// prioritised delivery of mutations that are needed by reads;
@ -71,7 +77,7 @@ public final class ActiveLogReconciler implements Shutdownable
*/
void schedule(ShortMutationId mutationId, InetAddressAndPort toHost, Priority priority)
{
queue(priority).offer(new Task(mutationId, toHost));
queue(priority).offer(Task.from(mutationId, toHost));
haveWork.release(1);
}
@ -82,7 +88,7 @@ public final class ActiveLogReconciler implements Shutdownable
void schedule(Offsets offsets, InetAddressAndPort toHost, Priority priority)
{
ManyToOneConcurrentLinkedQueue<Task> queue = queue(priority);
offsets.forEach(id -> queue.offer(new Task(id, toHost)));
offsets.forEach(id -> queue.offer(Task.from(id, toHost)));
haveWork.release(1);
}
@ -114,12 +120,26 @@ public final class ActiveLogReconciler implements Shutdownable
}
}
private static final class Task implements RequestCallback<NoPayload>
private static abstract class Task implements RequestCallback<NoPayload>
{
private static Task from(ShortMutationId id, InetAddressAndPort toHost)
{
CoordinatedTransfer transfer = LocalTransfers.instance().getActivatedTransfer(id);
if (transfer != null)
return new TransferTask(transfer, toHost);
else
return new MutationTask(id, toHost);
}
abstract void send();
}
private static final class MutationTask extends Task
{
private final ShortMutationId mutationId;
private final InetAddressAndPort toHost;
Task(ShortMutationId mutationId, InetAddressAndPort toHost)
MutationTask(ShortMutationId mutationId, InetAddressAndPort toHost)
{
this.mutationId = mutationId;
this.toHost = toHost;
@ -156,6 +176,59 @@ public final class ActiveLogReconciler implements Shutdownable
}
}
private static final class TransferTask extends Task
{
private final CoordinatedTransfer transfer;
private final InetAddressAndPort toHost;
TransferTask(CoordinatedTransfer transfer, InetAddressAndPort toHost)
{
this.transfer = transfer;
this.toHost = toHost;
}
@Override
public boolean invokeOnFailure()
{
return true;
}
@Override
public void onResponse(Message<NoPayload> msg)
{
logger.debug("Received activation ack for TransferTask from {}", toHost);
MutationTrackingService.instance.receivedActivationResponse(transfer, toHost);
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failureReason)
{
onFailure(failureReason.failure);
}
public void onFailure(Throwable cause)
{
logger.debug("Received activation failure for TransferTask from {} due to", toHost, cause);
MutationTrackingService.instance.retryFailedTransfer(transfer, toHost, cause);
}
void send()
{
logger.debug("Sending activation to {}", toHost);
LocalTransfers.instance().executor.submit(() -> {
try
{
transfer.activateOn(Collections.singleton(toHost));
onResponse(null);
}
catch (Throwable t)
{
onFailure(t);
}
});
}
}
private volatile boolean isShutdown = false;
private volatile boolean isPaused = false;

720
src/java/org/apache/cassandra/replication/CoordinatedTransfer.java Normal file
View File

@ -0,0 +1,720 @@
/*
* 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.replication;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import javax.annotation.CheckReturnValue;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.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.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.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeId;
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 org.apache.cassandra.utils.concurrent.Future;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.COMMITTED;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.COMMITTING;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.PREPARE_FAILED;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.PREPARING;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.INIT;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.STREAM_COMPLETE;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.STREAM_FAILED;
import static org.apache.cassandra.replication.CoordinatedTransfer.SingleTransferResult.State.STREAM_NOOP;
import static org.apache.cassandra.replication.TransferActivation.Phase;
/**
* Orchestrates the lifecycle of a tracked bulk data transfer for a single replica set, where the current instance is
* coordinating the transfer.
* <p>
* The transfer proceeds through these 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 sufficient replicas have received
* their streams to meet the requested {@link ConsistencyLevel}, the SSTables are activated using a two-phase
* commit protocol, making them part of the live set and visible to reads.
* </li>
* <li>
* <b>Activation {@link Phase#PREPARE}</b>
* The coordinator sends PREPARE messages to verify replicas have the data persisted on disk and are ready for
* activation.
* </li>
* <li>
* <b>Activation {@link Phase#COMMIT}</b>
* After successful PREPARE, the coordinator sends COMMIT messages to replicas. Replicas atomically move data from
* pending to live sets, making it visible to reads with the proper transfer ID in metadata. If commit succeeds
* on some replicas but not others, the transfer will be activated later on via existing the existing
* reconciliation processes (read reconciliation and background reconciliation).
* </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>
* If a tracked data read is executed on a replica that's missing an activation, the read reconciliation process will
* apply the missing activation during reconciliation and a subsequent read will succeed. To minimize the gap between
* activations across replicas, avoid expensive operations like file copies or index builds during
* {@link TransferActivation#apply()}.
*/
public class CoordinatedTransfer
{
private static final Logger logger = LoggerFactory.getLogger(CoordinatedTransfer.class);
String logPrefix()
{
return String.format("[CoordinatedTransfer #%s]", id);
}
private final ShortMutationId id;
private final String keyspace;
private final Range<Token> range;
private final ConsistencyLevel cl;
final Collection<SSTableReader> sstables;
final ConcurrentMap<InetAddressAndPort, SingleTransferResult> streamResults;
@VisibleForTesting
CoordinatedTransfer(Range<Token> range, MutationId id)
{
this.keyspace = null;
this.range = range;
this.sstables = Collections.emptyList();
this.cl = null;
this.id = id;
this.streamResults = new ConcurrentHashMap<>();
}
CoordinatedTransfer(String keyspace, Range<Token> range, Participants participants, Collection<SSTableReader> sstables, ConsistencyLevel cl, Supplier<MutationId> nextId)
{
this.keyspace = keyspace;
this.range = range;
this.sstables = sstables;
this.cl = cl;
this.id = nextId.get();
ClusterMetadata cm = ClusterMetadata.current();
this.streamResults = new ConcurrentHashMap<>(participants.size());
for (int i = 0; i < participants.size(); i++)
{
InetAddressAndPort addr = cm.directory.getNodeAddresses(new NodeId(participants.get(i))).broadcastAddress;
this.streamResults.put(addr, SingleTransferResult.Init());
}
}
ShortMutationId id()
{
return id;
}
void execute()
{
logger.debug("{} Executing tracked bulk transfer {}", logPrefix(), this);
LocalTransfers.instance().save(this);
stream();
}
private void stream()
{
// TODO: Don't stream multiple copies over the WAN, send one copy and indicate forwarding
List<Future<Void>> streaming = new ArrayList<>(streamResults.size());
for (InetAddressAndPort to : streamResults.keySet())
{
Future<Void> stream = LocalTransfers.instance().executor.submit(() -> {
stream(to);
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 boolean sufficient()
{
AbstractReplicationStrategy ars = Keyspace.open(keyspace).getReplicationStrategy();
int blockFor = cl.blockFor(ars);
int responses = 0;
for (Map.Entry<InetAddressAndPort, SingleTransferResult> entry : streamResults.entrySet())
{
if (entry.getValue().state == STREAM_COMPLETE)
responses++;
}
return responses >= blockFor;
}
void stream(InetAddressAndPort to)
{
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);
}
try
{
streamComplete(to, result);
}
catch (ExecutionException | InterruptedException | TimeoutException e)
{
Throwable cause = e instanceof ExecutionException ? e.getCause() : e;
throw Throwables.unchecked(cause);
}
}
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)
{
tryFailure(failure.failure);
}
}
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);
Message<TransferFailed> msg = Message.out(Verb.TRACKED_TRANSFER_FAILED_REQ, new TransferFailed(result.planId));
MessagingService.instance().sendWithCallback(msg, to, notifyFailure);
}
notifyFailure.get();
}
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));
}
/**
* This shouldn't throw an exception, even if we fail to notify peers of the streaming failure.
*/
private void maybeCleanupFailedStreams(Throwable cause)
{
try
{
boolean purgeable = LocalTransfers.instance().purger.test(this);
if (!purgeable)
return;
notifyFailure();
LocalTransfers.instance().scheduleCleanup();
}
catch (Throwable t)
{
if (cause != null)
t.addSuppressed(cause);
logger.error("{} Failed to notify peers of stream failure", logPrefix(), t);
}
}
private void streamComplete(InetAddressAndPort to, SingleTransferResult result) throws ExecutionException, InterruptedException, TimeoutException
{
streamResults.put(to, result);
logger.info("{} Completed streaming to {}, {}", logPrefix(), to, this);
maybeActivate();
}
synchronized void maybeActivate()
{
// If any activations have already been sent out, send new activations to any received plans that have not yet
// been activated
boolean anyActivated = false;
Set<InetAddressAndPort> awaitingActivation = new HashSet<>();
for (Map.Entry<InetAddressAndPort, SingleTransferResult> entry : streamResults.entrySet())
{
InetAddressAndPort peer = entry.getKey();
SingleTransferResult result = entry.getValue();
if (result.state == COMMITTING || result.state == COMMITTED)
{
anyActivated = true;
}
else if (result.state == STREAM_COMPLETE)
awaitingActivation.add(peer);
}
if (anyActivated && !awaitingActivation.isEmpty())
{
logger.debug("{} Transfer already activated on some peers, sending activations to remaining: {}", logPrefix(), awaitingActivation);
activateOn(awaitingActivation);
return;
}
// If no activations have been sent out, check whether we have enough planIds back to meet the required CL
else if (sufficient())
{
Set<InetAddressAndPort> peers = new HashSet<>();
for (Map.Entry<InetAddressAndPort, SingleTransferResult> entry : streamResults.entrySet())
{
InetAddressAndPort peer = entry.getKey();
SingleTransferResult result = entry.getValue();
if (result.state == STREAM_COMPLETE)
peers.add(peer);
}
logger.debug("{} Transfer meets consistency level {}, sending activations to {}", logPrefix(), cl, peers);
activateOn(peers);
return;
}
logger.debug("{} Nothing to activate", logPrefix());
}
void activateOn(Collection<InetAddressAndPort> peers)
{
Preconditions.checkState(!peers.isEmpty());
logger.debug("{} Activating {} on {}", logPrefix(), this, peers);
LocalTransfers.instance().activating(this);
// First phase ensures data is present on disk, then second phase does the actual import. This ensures that if
// something goes wrong (like a topology change during import), we don't have divergence.
class Prepare extends AsyncFuture<Void> implements RequestCallbackWithFailure<NoPayload>
{
final Set<InetAddressAndPort> responses = ConcurrentHashMap.newKeySet();
public Prepare()
{
responses.addAll(peers);
}
@Override
public void onResponse(Message<NoPayload> msg)
{
logger.debug("{} Got response from: {}", logPrefix(), msg.from());
responses.remove(msg.from());
if (responses.isEmpty())
trySuccess(null);
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
logger.debug("{} Got failure {} from {}", logPrefix(), failure, from);
CoordinatedTransfer.this.streamResults.computeIfPresent(from, (peer, result) -> result.prepareFailed());
tryFailure(new RuntimeException("Tracked import failed during PREPARE on " + from + " due to " + failure.reason));
}
}
Prepare prepare = new Prepare();
for (InetAddressAndPort peer : peers)
{
TransferActivation activation = new TransferActivation(this, peer, Phase.PREPARE);
Message<TransferActivation> msg = Message.out(Verb.TRACKED_TRANSFER_ACTIVATE_REQ, activation);
logger.debug("{} Sending {} to peer {}", logPrefix(), activation, peer);
MessagingService.instance().sendWithCallback(msg, peer, prepare);
CoordinatedTransfer.this.streamResults.computeIfPresent(peer, (peer0, result) -> result.preparing());
}
try
{
prepare.get();
}
catch (InterruptedException | ExecutionException e)
{
Throwable cause = e instanceof ExecutionException ? e.getCause() : e;
throw Throwables.unchecked(cause);
}
logger.debug("{} Activation prepare complete for {}", logPrefix(), peers);
// Acknowledgement of activation is equivalent to a remote write acknowledgement. The imported SSTables
// are now part of the live set, visible to reads.
class Commit extends AsyncFuture<Void> implements RequestCallbackWithFailure<Void>
{
final Set<InetAddressAndPort> responses = ConcurrentHashMap.newKeySet();
private Commit(Collection<InetAddressAndPort> peers)
{
responses.addAll(peers);
}
@Override
public void onResponse(Message<Void> msg)
{
logger.debug("{} Activation successfully applied on {}", logPrefix(), msg.from());
CoordinatedTransfer.this.streamResults.computeIfPresent(msg.from(), (peer, result) -> result.committed());
MutationTrackingService.instance.receivedActivationResponse(CoordinatedTransfer.this, msg.from());
responses.remove(msg.from());
if (responses.isEmpty())
{
// All activations complete, schedule cleanup to purge pending SSTables
LocalTransfers.instance().scheduleCleanup();
trySuccess(null);
}
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
logger.error("{} Failed activation on {} due to {}", logPrefix(), from, failure);
MutationTrackingService.instance.retryFailedTransfer(CoordinatedTransfer.this, from, failure.failure);
// TODO(expected): should only fail if we don't meet requested CL
tryFailure(new RuntimeException("Tracked import failed during COMMIT on " + from + " due to " + failure.reason));
}
}
Commit commit = new Commit(peers);
for (InetAddressAndPort peer : peers)
{
TransferActivation activation = new TransferActivation(this, peer, Phase.COMMIT);
Message<TransferActivation> msg = Message.out(Verb.TRACKED_TRANSFER_ACTIVATE_REQ, activation);
logger.debug("{} Sending {} to peer {}", logPrefix(), activation, peer);
MessagingService.instance().sendWithCallback(msg, peer, commit);
CoordinatedTransfer.this.streamResults.computeIfPresent(peer, (peer0, result) -> result.committing());
}
try
{
commit.get();
}
catch (InterruptedException | ExecutionException e)
{
Throwable cause = e instanceof ExecutionException ? e.getCause() : e;
throw Throwables.unchecked(cause);
}
logger.debug("{} Activation commit complete for {}", logPrefix(), peers);
}
public boolean isCommitted()
{
for (SingleTransferResult result : streamResults.values())
{
if (result.state != COMMITTED)
return false;
}
return true;
}
/**
* Tracks the lifecycle of a transfer from the coordinator to a single replica, using a two-phase commit protocol:
*
* <ul>
* <li>{@link State#INIT}: Transfer created, not yet streaming.</li>
* <li>{@link State#STREAM_COMPLETE}: Streaming successful, SSTables received on replica in pending directory.</li>
* <li>{@link State#STREAM_NOOP}: No data streamed (e.g., SSTable contains no rows in target range).</li>
* <li>{@link State#STREAM_FAILED}: Streaming failed, may not have a streaming plan ID yet.</li>
* <li>{@link State#PREPARING}: Preparing for activation (first phase).</li>
* <li>{@link State#PREPARE_FAILED}: Prepare failed, aborting transfer.</li>
* <li>{@link State#COMMITTING}: Committing transferred SSTables from pending to live set (second phase).</li>
* <li>{@link State#COMMITTED}: Transfer commit acknowledged on coordinator. SSTables now live and visible to reads.</li>
* </ul>
*
* <h3>Valid State Transitions:</h3>
* <pre>
* ┌────────────────┐
* ↓ │
* INIT ──┬──→ STREAM_COMPLETE ──→ PREPARING ──┬──→ COMMITTING ──→ COMMITTED
* │ │
* ├──→ STREAM_NOOP └──→ PREPARE_FAILED
* │
* └──→ STREAM_FAILED
* </pre>
*
* Failure states may be non-terminal if sufficient replicas reach successful states, depending on the transfer's
* consistency level.
*/
static class SingleTransferResult
{
enum State
{
INIT,
STREAM_NOOP,
STREAM_FAILED,
STREAM_COMPLETE,
PREPARING,
PREPARE_FAILED,
COMMITTING,
COMMITTED;
EnumSet<State> transitionFrom;
static
{
INIT.transitionFrom = EnumSet.noneOf(State.class);
STREAM_NOOP.transitionFrom = EnumSet.of(INIT);
STREAM_FAILED.transitionFrom = EnumSet.of(INIT);
STREAM_COMPLETE.transitionFrom = EnumSet.of(INIT);
PREPARING.transitionFrom = EnumSet.of(STREAM_COMPLETE, COMMITTING);
PREPARE_FAILED.transitionFrom = EnumSet.of(PREPARING);
COMMITTING.transitionFrom = EnumSet.of(PREPARING);
COMMITTED.transitionFrom = EnumSet.of(COMMITTING);
}
}
final State state;
private final TimeUUID planId;
@VisibleForTesting
SingleTransferResult(State state, TimeUUID planId)
{
this.state = state;
this.planId = planId;
}
private boolean canTransition(SingleTransferResult.State to)
{
return to.transitionFrom.contains(state);
}
public static SingleTransferResult Init()
{
return new SingleTransferResult(INIT, null);
}
@VisibleForTesting
static SingleTransferResult StreamComplete(TimeUUID planId)
{
return new SingleTransferResult(STREAM_COMPLETE, planId);
}
@VisibleForTesting
static SingleTransferResult Noop()
{
return new SingleTransferResult(STREAM_NOOP, null);
}
@CheckReturnValue
private SingleTransferResult transition(State to, TimeUUID planId)
{
if (!canTransition(to))
{
logger.error("Ignoring invalid transition from {} to {}", state, to);
return this;
}
// Don't overwrite if the stream succeeded but PREPARE failed, so we can clean up later
return new SingleTransferResult(to, planId == null ? this.planId : planId);
}
@CheckReturnValue
public SingleTransferResult streamFailed(TimeUUID planId)
{
return transition(STREAM_FAILED, planId);
}
@CheckReturnValue
public SingleTransferResult preparing()
{
return transition(PREPARING, this.planId);
}
@CheckReturnValue
public SingleTransferResult prepareFailed()
{
return transition(PREPARE_FAILED, this.planId);
}
@CheckReturnValue
public SingleTransferResult committing()
{
return transition(COMMITTING, this.planId);
}
@CheckReturnValue
public SingleTransferResult committed()
{
return transition(COMMITTED, this.planId);
}
public TimeUUID planId()
{
return planId;
}
@Override
public String toString()
{
return "SingleTransferResult{" +
"state=" + state +
", planId=" + planId +
'}';
}
}
private SingleTransferResult streamTask(InetAddressAndPort to) throws StreamException, ExecutionException, InterruptedException, TimeoutException
{
StreamPlan plan = new StreamPlan(StreamOperation.TRACKED_TRANSFER);
// No need to flush, only using non-live SSTables already on disk
plan.flushBeforeTransfer(false);
for (SSTableReader sstable : sstables)
{
List<Range<Token>> ranges = Collections.singletonList(range);
List<SSTableReader.PartitionPositionBounds> positions = sstable.getPositionsForRanges(ranges);
long estimatedKeys = sstable.estimatedKeysForRanges(ranges);
OutgoingStream stream = new CassandraOutgoingFile(StreamOperation.TRACKED_TRANSFER, 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());
}
@Override
public boolean equals(Object o)
{
if (o == null || getClass() != o.getClass()) return false;
CoordinatedTransfer transfer = (CoordinatedTransfer) o;
return Objects.equals(keyspace, transfer.keyspace) && Objects.equals(range, transfer.range) && Objects.equals(streamResults, transfer.streamResults) && Objects.equals(sstables, transfer.sstables) && cl == transfer.cl && Objects.equals(id, transfer.id);
}
@Override
public int hashCode()
{
return Objects.hash(keyspace, range, streamResults, sstables, cl, id);
}
@Override
public String toString()
{
return "CoordinatedTransfer{" +
"id=" + id +
", keyspace='" + keyspace + '\'' +
", range=" + range +
", cl=" + cl +
", sstables=" + sstables +
", streamResults=" + streamResults +
'}';
}
}

90
src/java/org/apache/cassandra/replication/CoordinatedTransfers.java Normal file
View File

@ -0,0 +1,90 @@
/*
* 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.replication;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.Interval;
/**
* Factory and container for creating multiple {@link CoordinatedTransfer} instances from a collection
* of SSTables, partitioned by {@link MutationTrackingService.KeyspaceShards}, which are aligned to replica ownership
* ranges. Each shard receives its own CoordinatedTransfer instance, which can be executed independently.
*/
class CoordinatedTransfers implements Iterable<CoordinatedTransfer>
{
private final Collection<CoordinatedTransfer> transfers;
private CoordinatedTransfers(Collection<CoordinatedTransfer> transfers)
{
this.transfers = transfers;
}
static CoordinatedTransfers create(String keyspace, MutationTrackingService.KeyspaceShards shards, Collection<SSTableReader> sstables, ConsistencyLevel cl)
{
// Clean up incoming SSTables to remove any existing untrusted CoordinatorLogOffsets
for (SSTableReader sstable : sstables)
{
try
{
sstable.mutateCoordinatorLogOffsetsAndReload(ImmutableCoordinatorLogOffsets.NONE);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
SSTableIntervalTree intervals = SSTableIntervalTree.buildSSTableIntervalTree(sstables);
List<CoordinatedTransfer> transfers = new ArrayList<>();
shards.forEachShard(shard -> {
Range<Token> range = shard.tokenRange();
Collection<SSTableReader> sstablesForRange = intervals.search(Interval.create(range.left.minKeyBound(), range.right.maxKeyBound()));
if (sstablesForRange.isEmpty())
return;
CoordinatedTransfer transfer = new CoordinatedTransfer(keyspace, range, shard.participants, sstablesForRange, cl, shard::nextId);
transfers.add(transfer);
});
return new CoordinatedTransfers(transfers);
}
@Override
public Iterator<CoordinatedTransfer> iterator()
{
return transfers.iterator();
}
@Override
public String toString()
{
return "CoordinatedTransfers{transfers=" + transfers + '}';
}
}

130
src/java/org/apache/cassandra/replication/CoordinatorLog.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.replication;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@ -148,37 +149,34 @@ public abstract class CoordinatorLog
// retroactively un-reconciling previously reconciled offsets for the other replicas.
offsets.addAll(reconciledOffsets);
}
Offsets.Mutable persisted = participants.contains(participantId)
? persistedOffsets.get(participantId)
: new Offsets.Mutable(logId);
passivelyReconciled = passivelyReconciled != null
? Offsets.Immutable.intersection(passivelyReconciled, offsets)
: offsets;
newWitnessedOffsets.add(participantId, offsets);
newPersistedOffsets.add(participantId, persisted);
}
UnreconciledMutations newUnreconciled;
UnreconciledMutations newUnreconciledMutations;
passivelyReconciled = Offsets.Immutable.difference(passivelyReconciled, reconciledOffsets);
if (!passivelyReconciled.isEmpty())
{
logger.debug("Toplogy change implicitly reconciled offsets: {}", passivelyReconciled);
newUnreconciled = unreconciledMutations.copy();
passivelyReconciled.forEach(id -> newUnreconciled.remove(id.offset));
newUnreconciledMutations = unreconciledMutations.copy();
passivelyReconciled.forEach(id -> newUnreconciledMutations.remove(id.offset));
}
else
{
newUnreconciled = unreconciledMutations;
newUnreconciledMutations = unreconciledMutations;
}
if (logger.isTraceEnabled())
logger.trace("Updating coordinator log {} participants: {} -> {}. Passively reconciled: {}",
logId, participants, newParticipants, passivelyReconciled);
return withUpdatedParticipants(newParticipants, newWitnessedOffsets, newPersistedOffsets, newUnreconciled);
return withUpdatedParticipants(newParticipants, newWitnessedOffsets, newPersistedOffsets, newUnreconciledMutations);
}
finally
{
@ -218,6 +216,7 @@ public abstract class CoordinatorLog
reconciledOffsets.add(offset);
unreconciledMutations.remove(offset);
}
logger.trace("done applying WRO, now {}", witnessedOffsets);
}
});
}
@ -225,7 +224,9 @@ public abstract class CoordinatorLog
private void updatePersistedReplicatedOffsets(Offsets offsets, int onNodeId)
{
persistedOffsets.get(onNodeId).addAll(offsets);
logger.debug("done applying PO, now {}", persistedOffsets);
reconciledPersistedOffsets.addAll(persistedOffsets.intersection());
logger.debug("done applying PRO, now {}", reconciledPersistedOffsets);
}
public void recordFullyReconciledOffsets(Offsets.Immutable reconciled)
@ -331,17 +332,80 @@ public abstract class CoordinatorLog
return othersWitnessed(offset, localNodeId);
}
/*
- On local replicas after they've completed activation (onHostId == me)
*/
void finishActivation(PendingLocalTransfer transfer, TransferActivation activation)
{
logger.trace("witnessed local transfer {}", activation.id());
lock.writeLock().lock();
try
{
int offset = activation.id().offset();
// we've raced with another write, no need to do anything else
if (!witnessedOffsets.get(localNodeId).add(offset))
return;
// This is the only difference with finishWriting - can we consolidate these methods?
unreconciledMutations.activatedTransfer(activation.id(), transfer.sstables);
if (remoteReplicasWitnessed(offset))
{
reconciledOffsets.add(offset);
unreconciledMutations.remove(offset);
}
}
finally
{
lock.writeLock().unlock();
}
}
/*
- On transfer coordinators after they've received a completed activation from a peer (onHostId != me)
- On local replicas after coordinators have propagated their replicated offsets
*/
void receivedActivationResponse(CoordinatedTransfer transfer, int onHostId)
{
ShortMutationId transferId = transfer.id();
Preconditions.checkArgument(!transferId.isNone());
logger.trace("witnessed transfer activation ack {} from {}", transferId, onHostId);
lock.writeLock().lock();
try
{
if (!witnessedOffsets.get(onHostId).add(transferId.offset()))
return; // already witnessed; very uncommon but possible path
if (!witnessedOffsets.get(localNodeId).contains(transferId.offset()))
return; // local host hasn't witnessed yet -> no cleanup needed
if (remoteReplicasWitnessed(transferId.offset()))
{
logger.trace("marking transfer {} as fully reconciled", transferId);
// if all replicas have now witnessed the id, remove it from the index
unreconciledMutations.remove(transferId.offset());
reconciledOffsets.add(transferId.offset());
}
}
finally
{
logger.trace("after receivedActivationAck {} witnessed by: {}", transferId, witnessedOffsets);
lock.writeLock().unlock();
}
}
/**
* Look up unreconciled sequence ids of mutations witnessed by this host in this coordinataor log.
* Adds the ids to the supplied collection, so it can be reused to aggregate lookups for multiple logs.
*/
boolean collectOffsetsFor(Token token, TableId tableId, boolean includePending, Offsets.OffsetReciever unreconciledInto, Offsets.OffsetReciever reconciledInto)
void collectOffsetsFor(Token token, TableId tableId, boolean includePending, Offsets.OffsetReciever unreconciledInto, Offsets.OffsetReciever reconciledInto)
{
lock.readLock().lock();
try
{
reconciledInto.addAll(reconciledOffsets);
return unreconciledMutations.collect(token, tableId, includePending, unreconciledInto);
unreconciledMutations.collect(token, tableId, includePending, unreconciledInto);
}
finally
{
@ -353,13 +417,13 @@ public abstract class CoordinatorLog
* Look up unreconciled sequence ids of mutations witnessed by this host in this coordinataor log.
* Adds the ids to the supplied collection, so it can be reused to aggregate lookups for multiple logs.
*/
boolean collectOffsetsFor(AbstractBounds<PartitionPosition> range, TableId tableId, boolean includePending, Offsets.OffsetReciever unreconciledInto, Offsets.OffsetReciever reconciledInto)
void collectOffsetsFor(AbstractBounds<PartitionPosition> range, TableId tableId, boolean includePending, Offsets.OffsetReciever unreconciledInto, Offsets.OffsetReciever reconciledInto)
{
lock.readLock().lock();
try
{
reconciledInto.addAll(reconciledOffsets);
return unreconciledMutations.collect(range, tableId, includePending, unreconciledInto);
unreconciledMutations.collect(range, tableId, includePending, unreconciledInto);
}
finally
{
@ -406,14 +470,50 @@ public abstract class CoordinatorLog
into.add(reconciledPersistedOffsets);
}
boolean isDurablyReconciled(ShortMutationId id)
{
lock.readLock().lock();
try
{
boolean contains = reconciledPersistedOffsets.contains(id.offset);
if (!contains)
logger.debug("Offset {} is not contained in durably reconciled offsets {}", id.offset, reconciledPersistedOffsets);
return contains;
}
finally
{
lock.readLock().unlock();
}
}
private boolean isDurablyReconciled(Iterator<? extends ShortMutationId> ids)
{
if (ids == null)
return true;
while (ids.hasNext())
{
ShortMutationId id = ids.next();
if (id.logId() != logId.asLong())
continue;
if (!isDurablyReconciled(id))
return false;
}
return true;
}
boolean isDurablyReconciled(CoordinatorLogOffsets<?> logOffsets)
{
lock.readLock().lock();
try
{
Offsets.RangeIterator durablyReconciled = reconciledPersistedOffsets.rangeIterator();
Offsets.RangeIterator difference = Offsets.difference(logOffsets.offsets(logId.asLong()).rangeIterator(), durablyReconciled);
return !difference.tryAdvance();
// Mutations only
Offsets.RangeIterator offsets = logOffsets.mutations().offsets(logId.asLong()).rangeIterator();
Offsets.RangeIterator unreconciledMutations = Offsets.difference(offsets, durablyReconciled);
// Transfers
boolean transfersReconciled = isDurablyReconciled(logOffsets.transfers().iterator());
return transfersReconciled && !unreconciledMutations.tryAdvance();
}
finally
{
@ -447,8 +547,6 @@ public abstract class CoordinatorLog
super(keyspace, range, localNodeId, logId, participants);
}
@Override
CoordinatorLog withUpdatedParticipants(Participants newParticipants, Node2OffsetsMap witnessedOffsets, Node2OffsetsMap persistedOffsets, UnreconciledMutations unreconciledMutations)
{

29
src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java
View File

@ -22,18 +22,35 @@ import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
/**
* Mutation ID offsets present in this SSTable for each coordinator log, to determine whether an SSTable is reconciled
* or not.
* or not. This includes "plain" mutation IDs for regular writes, and "activation" mutation IDs for bulk transfers.
* Bulk transfer IDs are kept separately because we expect to have very few of them, and they're materialized for fast
* access on the read path.
* <p>
* Note that peers may have reconciled all mutations included in an SSTable, but {@link StatsMetadata#repairedAt} is
* dependent on compaction timing, so "nodetool repair --validate" may report temporary disagreements on the repaired
* set.
* <p>
* Iterable over {@link CoordinatorLogId}.
*/
public interface CoordinatorLogOffsets<O extends Offsets> extends Iterable<Long>
public interface CoordinatorLogOffsets<O extends Offsets>
{
O offsets(long logId);
int size();
/**
* Iterable over {@link CoordinatorLogId}.
*/
interface Mutations<O> extends Iterable<Long>
{
O offsets(long logId);
int size();
default boolean isEmpty()
{
return size() == 0;
}
}
Mutations<O> mutations();
default ActivatedTransfers transfers()
{
return ActivatedTransfers.EMPTY;
}
ImmutableCoordinatorLogOffsets NONE = new ImmutableCoordinatorLogOffsets.Builder(0).build();
}

2
src/java/org/apache/cassandra/replication/ForwardedWrite.java
View File

@ -204,7 +204,7 @@ public class ForwardedWrite
ClusterMetadata cm = ClusterMetadata.current();
Token token = mutation.key().getToken();
Keyspace keyspace = Keyspace.open(mutation.getKeyspaceName());
EndpointsForRange endpoints = cm.placements.get(keyspace.getMetadata().params.replication).writes.forRange(token).get();
EndpointsForToken endpoints = cm.placements.get(keyspace.getMetadata().params.replication).writes.forToken(token).get();
logger.trace("Finding best leader from replicas {}", endpoints);
// TODO: Should match ReplicaPlans.findCounterLeaderReplica, including DC-local priority, current health, severity, etc.

257
src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
View File

@ -19,16 +19,24 @@
package org.apache.cassandra.replication;
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.function.BiConsumer;
import java.util.function.Predicate;
import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterators;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.agrona.collections.Long2ObjectHashMap;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
@ -36,73 +44,73 @@ import org.apache.cassandra.utils.vint.VIntCoding;
public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Offsets.Immutable>
{
private final Long2ObjectHashMap<Offsets.Immutable> ids;
private static final Logger logger = LoggerFactory.getLogger(ImmutableCoordinatorLogOffsets.class);
@Override
public Offsets.Immutable offsets(long logId)
private final ImmutableMutations mutations;
private final ActivatedTransfers transfers;
private ImmutableCoordinatorLogOffsets(Builder builder)
{
Offsets.Immutable offsets = ids.get(logId);
if (offsets == null)
return new Offsets.Immutable(new CoordinatorLogId(logId));
return offsets;
// Important to set shouldAvoidAllocation=false, otherwise iterators are cached and not thread safe, even when immutable and read-only
Long2ObjectHashMap<Offsets.Immutable> ids = new Long2ObjectHashMap<>(builder.ids.size(), 0.9f, false);
for (Map.Entry<Long, Offsets.Immutable.Builder> entry : builder.ids.entrySet())
ids.put(entry.getKey(), entry.getValue().build());
this.mutations = new ImmutableMutations(ids);
this.transfers = ActivatedTransfers.copyOf(builder.transfers);
}
@Override
public int size()
public Mutations<Offsets.Immutable> mutations()
{
return ids.size();
}
public boolean isEmpty()
{
return size() == 0;
return mutations;
}
@Override
public Iterator<Long> iterator()
public ActivatedTransfers transfers()
{
return Iterators.unmodifiableIterator(ids.keySet().iterator());
return transfers == null ? ActivatedTransfers.EMPTY : transfers;
}
public Iterable<Map.Entry<Long, Offsets.Immutable>> entries()
{
return ids.entrySet();
return mutations.ids.entrySet();
}
public boolean isEmpty()
{
return mutations().isEmpty() && transfers().isEmpty();
}
@Override
public boolean equals(Object o)
{
if (o == null || getClass() != o.getClass()) return false;
ImmutableCoordinatorLogOffsets longs = (ImmutableCoordinatorLogOffsets) o;
return Objects.equals(ids, longs.ids);
ImmutableCoordinatorLogOffsets other = (ImmutableCoordinatorLogOffsets) o;
return Objects.equals(mutations, other.mutations) && Objects.equals(transfers, other.transfers);
}
@Override
public int hashCode()
{
return Objects.hashCode(ids);
return Objects.hash(mutations, transfers);
}
public ImmutableCoordinatorLogOffsets(Builder builder)
@Override
public String toString()
{
// Important to set shouldAvoidAllocation=false, otherwise iterators are cached and not thread safe, even when
// immutable and read-only
this.ids = new Long2ObjectHashMap<>(builder.ids.size(), 0.9f, false);
for (Map.Entry<Long, Offsets.Immutable.Builder> entry : builder.ids.entrySet())
ids.put(entry.getKey(), entry.getValue().build());
}
public void forEach(BiConsumer<CoordinatorLogId, Offsets.Immutable> consumer)
{
ids.forEach((logId, offsets) -> consumer.accept(new CoordinatorLogId(logId), offsets));
return "ImmutableCoordinatorLogOffsets{" +
"mutations=" + mutations +
", transfers=" + transfers +
'}';
}
@NotThreadSafe
public static class Builder
{
private final Long2ObjectHashMap<Offsets.Immutable.Builder> ids;
private ActivatedTransfers transfers;
public Builder()
{
@ -112,6 +120,7 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off
public Builder(int size)
{
this.ids = new Long2ObjectHashMap<>(size, 0.9f, false);
this.transfers = null;
}
public Builder add(MutationId mutationId)
@ -123,17 +132,28 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off
return this;
}
public Builder addAll(CoordinatorLogOffsets<?> logOffsets)
private Builder addAll(CoordinatorLogOffsets.Mutations<? extends Offsets> mutations)
{
for (long log : logOffsets)
for (long log : mutations)
{
Offsets offsets = logOffsets.offsets(log);
Offsets offsets = mutations.offsets(log);
ids.computeIfAbsent(log, logId -> new Offsets.Immutable.Builder(new CoordinatorLogId(logId)))
.addAll(offsets);
}
return this;
}
public Builder addAll(CoordinatorLogOffsets<?> logOffsets)
{
addAll(logOffsets.mutations());
ActivatedTransfers newTransfers = logOffsets.transfers();
if (transfers == null)
transfers = newTransfers;
else
transfers.addAll(newTransfers);
return this;
}
public Builder addAll(Offsets.Immutable offsets)
{
ids.computeIfAbsent(offsets.logId.asLong(), logId -> new Offsets.Immutable.Builder(new CoordinatorLogId(logId)))
@ -141,6 +161,62 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off
return this;
}
@VisibleForTesting
public Builder addTransfer(ShortMutationId transferId, Bounds<Token> bounds)
{
if (transferId.isNone())
return this;
if (transfers == null)
transfers = new ActivatedTransfers();
transfers.add(transferId, bounds);
return this;
}
public Builder addTransfer(ShortMutationId transferId, Collection<SSTableReader> sstables)
{
if (transferId.isNone())
return this;
if (transfers == null)
transfers = new ActivatedTransfers();
transfers.add(transferId, sstables);
return this;
}
public Builder addTransfers(ActivatedTransfers other)
{
if (other.isEmpty())
return this;
if (transfers == null)
transfers = other;
else
transfers.addAll(other);
return this;
}
/**
* Removes expired transfers
*/
public void purgeTransfers(Predicate<ShortMutationId> predicate)
{
int purged = 0;
if (transfers != null)
{
Iterator<ShortMutationId> iter = transfers.iterator();
while (iter.hasNext())
{
ShortMutationId id = iter.next();
if (predicate.test(id))
{
iter.remove();
purged++;
logger.debug("Purging activation {}", id);
}
}
}
if (purged > 0)
logger.info("Purged {} transfers", purged);
}
public ImmutableCoordinatorLogOffsets build()
{
return new ImmutableCoordinatorLogOffsets(this);
@ -154,9 +230,8 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off
{
if (version < MessagingService.VERSION_61)
return;
out.writeUnsignedVInt32(logOffsets.size());
for (long logId : logOffsets)
Offsets.serializer.serialize(logOffsets.offsets(logId), out, version);
ImmutableMutations.serializer.serialize(logOffsets.mutations, out, version);
ActivatedTransfers.serializer.serialize(logOffsets.transfers(), out, version);
}
@Override
@ -164,13 +239,12 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off
{
if (version < MessagingService.VERSION_61)
return ImmutableCoordinatorLogOffsets.NONE;
int size = in.readUnsignedVInt32();
ImmutableCoordinatorLogOffsets.Builder builder = new ImmutableCoordinatorLogOffsets.Builder(size);
for (int i = 0; i < size; i++)
{
Offsets.Immutable offsets = Offsets.serializer.deserialize(in, version);
builder.addAll(offsets);
}
Builder builder = new Builder();
ImmutableMutations mutations = ImmutableMutations.serializer.deserialize(in, version);
mutations.ids.forEach((id, offsets) -> builder.addAll(offsets));
ActivatedTransfers transfers = ActivatedTransfers.serializer.deserialize(in, version);
if (!transfers.isEmpty())
builder.addTransfers(transfers);
return builder.build();
}
@ -180,12 +254,97 @@ public class ImmutableCoordinatorLogOffsets implements CoordinatorLogOffsets<Off
if (version < MessagingService.VERSION_61)
return 0;
long size = 0;
size += VIntCoding.computeUnsignedVIntSize(logOffsets.size());
for (long logId : logOffsets)
size += Offsets.serializer.serializedSize(logOffsets.offsets(logId), version);
size += ImmutableMutations.serializer.serializedSize(logOffsets.mutations, version);
size += ActivatedTransfers.serializer.serializedSize(logOffsets.transfers(), version);
return size;
}
}
public static final Serializer serializer = new Serializer();
public static class ImmutableMutations implements Mutations<Offsets.Immutable>
{
final private Long2ObjectHashMap<Offsets.Immutable> ids;
private ImmutableMutations(Long2ObjectHashMap<Offsets.Immutable> ids)
{
this.ids = ids;
}
@Override
public Offsets.Immutable offsets(long logId)
{
Offsets.Immutable offsets = ids.get(logId);
if (offsets == null)
return new Offsets.Immutable(new CoordinatorLogId(logId));
return offsets;
}
@Override
public int size()
{
return ids.size();
}
@Override
public Iterator<Long> iterator()
{
return Iterators.unmodifiableIterator(ids.keySet().iterator());
}
@Override
public boolean equals(Object o)
{
if (o == null || getClass() != o.getClass()) return false;
ImmutableMutations longs = (ImmutableMutations) o;
return Objects.equals(ids, longs.ids);
}
@Override
public int hashCode()
{
return Objects.hashCode(ids);
}
@Override
public String toString()
{
return "ImmutableMutations{" +
"ids=" + ids +
'}';
}
private static final IVersionedSerializer<ImmutableMutations> serializer = new IVersionedSerializer<>()
{
@Override
public void serialize(ImmutableMutations mutations, DataOutputPlus out, int version) throws IOException
{
out.writeUnsignedVInt32(mutations.size());
for (long logId : mutations)
Offsets.serializer.serialize(mutations.offsets(logId), out, version);
}
@Override
public ImmutableMutations deserialize(DataInputPlus in, int version) throws IOException
{
int size = in.readUnsignedVInt32();
Long2ObjectHashMap<Offsets.Immutable> ids = new Long2ObjectHashMap<>(size, 0.9f, false);
for (int i = 0; i < size; i++)
{
Offsets.Immutable offsets = Offsets.serializer.deserialize(in, version);
ids.put(offsets.logId.asLong(), offsets);
}
return new ImmutableMutations(ids);
}
@Override
public long serializedSize(ImmutableMutations mutations, int version)
{
long size = VIntCoding.computeUnsignedVIntSize(mutations.size());
for (long logId : mutations)
size += Offsets.serializer.serializedSize(mutations.offsets(logId), version);
return size;
}
};
}
}

290
src/java/org/apache/cassandra/replication/LocalTransfers.java Normal file
View File

@ -0,0 +1,290 @@
/*
* 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.replication;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.annotation.Nullable;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.io.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.utils.FBUtilities;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
/**
* Singleton registry maintaining state for bulk data transfers on the local node.
* <p>
* This includes {@link CoordinatedTransfer} instances that the current node is coordinating, and
* {@link PendingLocalTransfer} instances that are coordinated by other nodes. Pending transfers are inactive until
* activated by the coordinator.
* <p>
* TODO: Make changes to pending set durable with SystemKeyspace.savePendingLocalTransfer(transfer)?
* TODO: Add vtable for visibility into local and coordinated transfers
*/
public class LocalTransfers
{
private static final Logger logger = LoggerFactory.getLogger(LocalTransfers.class);
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final Map<ShortMutationId, CoordinatedTransfer> coordinating = new ConcurrentHashMap<>();
private final Map<TimeUUID, PendingLocalTransfer> local = new ConcurrentHashMap<>();
final ExecutorPlus executor = executorFactory().pooled("LocalTrackedTransfers", Integer.MAX_VALUE);
private 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 activating(CoordinatedTransfer transfer)
{
Preconditions.checkNotNull(transfer.id());
lock.writeLock().lock();
try
{
coordinating.put(transfer.id(), transfer);
}
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();
}
}
Purger purger = new Purger();
static class Purger
{
/**
* It's safe to purge a transfer if it failed either before it was activated anywhere, or after all activation
* has completed everywhere. If a transfer is partially activated (on some replicas but not others), it's going
* to be included in future reconciliations and needs to be preserved until reconciliation is complete.
*/
boolean test(CoordinatedTransfer transfer)
{
logger.debug("Checking whether we can purge {}", transfer);
boolean failedBeforeActivation = false;
boolean noneActivated = true;
boolean allComplete = true;
for (CoordinatedTransfer.SingleTransferResult result : transfer.streamResults.values())
{
switch (result.state)
{
case STREAM_FAILED:
case PREPARE_FAILED:
failedBeforeActivation = true;
break;
case COMMITTED:
noneActivated = false;
break;
case INIT:
case STREAM_COMPLETE:
case PREPARING:
case COMMITTING:
allComplete = false;
}
}
return (failedBeforeActivation && noneActivated) || allComplete;
}
boolean test(PendingLocalTransfer transfer)
{
return transfer.activated;
}
}
private void cleanup()
{
lock.writeLock().lock();
try
{
for (PendingLocalTransfer transfer : local.values())
if (purger.test(transfer))
purge(transfer);
for (CoordinatedTransfer transfer : coordinating.values())
if (purger.test(transfer))
purge(transfer);
}
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());
if (transfer.id() != null)
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 scheduleCleanup()
{
executor.submit(() -> {
try
{
cleanup();
}
catch (Throwable t)
{
logger.error("Cleanup failed", t);
}
});
}
@Nullable PendingLocalTransfer getPendingTransfer(TimeUUID planId)
{
lock.readLock().lock();
try
{
return local.get(planId);
}
finally
{
lock.readLock().unlock();
}
}
@Nullable CoordinatedTransfer getActivatedTransfer(ShortMutationId transferId)
{
lock.readLock().lock();
try
{
return coordinating.get(transferId);
}
finally
{
lock.readLock().unlock();
}
}
public static IVerbHandler<TransferFailed> verbHandler = message -> {
LocalTransfers.instance().purge(message.payload);
MessagingService.instance().respond(NoPayload.noPayload, message);
};
}

6
src/java/org/apache/cassandra/replication/MutableCoordinatorLogOffsets.java
View File

@ -28,11 +28,11 @@ public interface MutableCoordinatorLogOffsets extends CoordinatorLogOffsets<Offs
{
void add(ShortMutationId mutationId);
default void addAll(ImmutableCoordinatorLogOffsets from)
default void addAll(Mutations<? extends Offsets> mutations)
{
for (long logId : from)
for (long logId : mutations)
{
Offsets offsets = from.offsets(logId);
Offsets offsets = mutations.offsets(logId);
offsets.forEach(this::add);
}
}

129
src/java/org/apache/cassandra/replication/MutationTrackingService.java
View File

@ -38,10 +38,12 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import org.agrona.collections.IntArrayList;
import org.agrona.collections.IntHashSet;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Mutation;
@ -53,6 +55,7 @@ import org.apache.cassandra.dht.Splitter;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
@ -75,6 +78,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.lang.String.format;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.concurrent.ExecutorFactory.SimulatorSemantics.NORMAL;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
@ -88,9 +92,12 @@ public class MutationTrackingService
/**
* Split ranges into this many shards.
* <p>
* REVIEW: Reset back to 1 because for transfers, replicas need to know each others' shards, since transfers are
* sliced to fit within shards. Can we achieve sharding via split range ownership, instead of it being local-only?
* <p>
* TODO (expected): ability to rebalance / change this constant
*/
private static final int SHARD_MULTIPLIER = 8;
private static final int SHARD_MULTIPLIER = 1;
private static final Logger logger = LoggerFactory.getLogger(MutationTrackingService.class);
public static final MutationTrackingService instance = new MutationTrackingService();
@ -250,6 +257,28 @@ public class MutationTrackingService
}
}
public void receivedActivationResponse(CoordinatedTransfer transfer, InetAddressAndPort fromHost)
{
shardLock.readLock().lock();
try
{
logger.debug("{} receivedActivationAck from {}", transfer.logPrefix(), fromHost);
Preconditions.checkArgument(!transfer.id().isNone());
// REVIEW: This will be called with ShortMutationId, which overrides hashCode from CoordinatorLogId, but map
// is updated with CoordinatorLogId; shouldn't call this with a ShortMutationId, not sure why that's working
// elsewhere
Shard shard = getShardNullable(new CoordinatorLogId(transfer.id().logId()));
// Local activation acknowledged in MutationTrackingService.activateLocal
if (shard != null && !fromHost.equals(FBUtilities.getBroadcastAddressAndPort()))
shard.receivedActivationResponse(transfer, fromHost);
}
finally
{
shardLock.readLock().unlock();
}
}
public void retryFailedWrite(ShortMutationId mutationId, InetAddressAndPort onHost, RequestFailure reason)
{
Preconditions.checkArgument(!mutationId.isNone());
@ -257,6 +286,18 @@ public class MutationTrackingService
activeReconciler.schedule(mutationId, onHost, ActiveLogReconciler.Priority.REGULAR);
}
public void retryFailedTransfer(CoordinatedTransfer transfer, InetAddressAndPort onHost, Throwable cause)
{
if (transfer.isCommitted())
{
logger.debug("Failed transfer {} to {} is already committed, skipping reconciliation", transfer, onHost, cause);
return;
}
logger.debug("Retrying failed transfer {} to {} with exception", transfer, onHost, cause);
Preconditions.checkArgument(!transfer.id().isNone());
activeReconciler.schedule(transfer.id(), onHost, ActiveLogReconciler.Priority.REGULAR);
}
public void updateReplicatedOffsets(String keyspace, Range<Token> range, List<? extends Offsets> offsets, boolean durable, InetAddressAndPort onHost)
{
shardLock.readLock().lock();
@ -325,6 +366,54 @@ public class MutationTrackingService
return incomingMutations.subscribe(mutationId, callback);
}
public void executeTransfers(String keyspace, Set<SSTableReader> sstables, ConsistencyLevel cl)
{
shardLock.readLock().lock();
try
{
logger.info("Creating tracked bulk transfers for keyspace '{}' SSTables {}...", keyspace, sstables);
KeyspaceShards shards = checkNotNull(keyspaceShards.get(keyspace));
CoordinatedTransfers transfers = CoordinatedTransfers.create(keyspace, shards, sstables, cl);
logger.info("Split input SSTables into transfers {}", transfers);
for (CoordinatedTransfer transfer : transfers)
transfer.execute();
}
finally
{
shardLock.readLock().unlock();
}
}
public void received(PendingLocalTransfer transfer)
{
logger.debug("Received pending transfer for tracked table {}", transfer);
LocalTransfers.instance().received(transfer);
}
void activateLocal(TransferActivation activation)
{
PendingLocalTransfer pending = LocalTransfers.instance().getPendingTransfer(activation.planId);
if (pending == null)
throw new IllegalStateException(String.format("Cannot activate unknown local pending transfer %s", activation));
pending.activate(activation);
shardLock.readLock().lock();
try
{
if (activation.isCommit())
{
keyspaceShards.get(pending.keyspace).lookUp(pending.range).finishActivation(pending, activation);
incomingMutations.invokeListeners(activation.transferId);
}
}
finally
{
shardLock.readLock().unlock();
}
}
public MutationSummary createSummaryForKey(DecoratedKey key, TableId tableId, boolean includePending)
{
shardLock.readLock().lock();
@ -459,14 +548,33 @@ public class MutationTrackingService
}
private int prevHostLogId;
public boolean isDurablyReconciled(ShortMutationId id)
{
shardLock.readLock().lock();
try
{
long logId = id.logId();
Shard shard = getShardNullable(new CoordinatorLogId(logId));
if (shard == null)
throw new IllegalStateException("Could not find shard for logId " + logId);
return shard.isDurablyReconciled(id);
}
finally
{
shardLock.readLock().unlock();
}
}
public boolean isDurablyReconciled(ImmutableCoordinatorLogOffsets logOffsets)
{
shardLock.readLock().lock();
try
{
// Could pass through SSTable bounds to exclude shards for non-overlapping ranges, but this will mostly be
// called on flush for L0 SSTables with wide bounds.
for (Long logId : logOffsets)
Iterable<Long> mutations = logOffsets.mutations();
Iterable<Long> transfers = Iterables.transform(logOffsets.transfers(), ShortMutationId::logId);
Iterable<Long> logIds = Iterables.concat(mutations, transfers);
for (Long logId : logIds)
{
Shard shard = getShardNullable(new CoordinatorLogId(logId));
if (shard == null)
@ -886,7 +994,18 @@ public class MutationTrackingService
Shard lookUp(Token token)
{
return shards.get(groups.forRange(token).range());
VersionedEndpoints.ForRange forRange = groups.matchToken(token);
if (forRange == null)
throw new UnknownShardException(token, groups);
return shards.get(forRange.range());
}
Shard lookUp(Range<Token> range)
{
VersionedEndpoints.ForRange forRange = groups.matchRange(range);
if (forRange == null)
throw new UnknownShardException(range, groups);
return shards.get(forRange.range());
}
void persistToSystemTables()

8
src/java/org/apache/cassandra/replication/Node2OffsetsMap.java
View File

@ -122,4 +122,12 @@ public class Node2OffsetsMap
Node2OffsetsMap that = (Node2OffsetsMap) o;
return this.offsetsMap.equals(that.offsetsMap);
}
@Override
public String toString()
{
return "Node2OffsetsMap{" +
"offsetsMap=" + offsetsMap +
'}';
}
}

33
src/java/org/apache/cassandra/replication/NonBlockingCoordinatorLogOffsets.java
View File

@ -71,18 +71,31 @@ abstract class NonBlockingCoordinatorLogOffsets<E extends NonBlockingCoordinator
}
@Override
public Offsets.Mutable offsets(long logId)
public Mutations<Offsets.Mutable> mutations()
{
E logOffsets = get(logId);
if (logOffsets == null)
return new Offsets.Mutable(new CoordinatorLogId(logId));
return logOffsets.offsets();
}
return new Mutations<>()
{
@Override
public Offsets.Mutable offsets(long logId)
{
E logOffsets = get(logId);
if (logOffsets == null)
return new Offsets.Mutable(new CoordinatorLogId(logId));
return logOffsets.offsets();
}
@Override
public Iterator<Long> iterator()
{
return Iterators.unmodifiableIterator(keys().asIterator());
@Override
public int size()
{
return NonBlockingCoordinatorLogOffsets.super.size();
}
@Override
public Iterator<Long> iterator()
{
return Iterators.unmodifiableIterator(keys().asIterator());
}
};
}
public static class Exclusive extends NonBlockingCoordinatorLogOffsets<Exclusive.Entry>

243
src/java/org/apache/cassandra/replication/PendingLocalTransfer.java Normal file
View File

@ -0,0 +1,243 @@
/*
* 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.replication;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Objects;
import java.util.function.Consumer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ownership.ReplicaGroups;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
/**
* Represents a bulk data transfer received on a replica, from completion of streaming into the pending location,
* through activation when it's made visible to reads. Pending transfers are identified by their streaming plan ID,
* and made live by {@link TransferActivation} which associates the streaming plan with a transfer ID that can be
* represented in mutation summaries.
*/
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;
transient Range<Token> range;
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();
this.range = shardRange(keyspace, sstables);
}
@VisibleForTesting
PendingLocalTransfer(TimeUUID planId, Collection<SSTableReader> sstables)
{
Preconditions.checkState(!sstables.isEmpty());
this.planId = planId;
this.tableId = null;
this.sstables = sstables;
this.keyspace = null;
this.range = null;
}
/**
* Pending transfers should be within a single shard, which are aligned to natural ranges.
* See ({@link MutationTrackingService.KeyspaceShards#make}).
*/
private static Range<Token> shardRange(String keyspace, Collection<SSTableReader> sstables)
{
ClusterMetadata cm = ClusterMetadata.current();
ReplicaGroups writes = cm.placements.get(Keyspace.open(keyspace).getMetadata().params.replication).writes;
Range<Token> range = null;
for (SSTableReader sstable : sstables)
{
if (range == null)
{
Token first = sstable.getFirst().getToken();
range = writes.forRange(first).range();
}
else
{
AbstractBounds<Token> bounds = sstable.getBounds();
Preconditions.checkState(!range.isTrulyWrapAround());
Preconditions.checkState(range.contains(bounds.left));
Preconditions.checkState(range.contains(bounds.right));
}
}
Preconditions.checkNotNull(range);
return range;
}
private boolean isFullReplica()
{
ClusterMetadata cm = ClusterMetadata.current();
Keyspace ks = Keyspace.open(keyspace);
ReplicaGroups writes = cm.placements.get(ks.getMetadata().params.replication).writes;
EndpointsForRange replicas = writes.forRange(range.right).get();
return replicas.containsSelf() && replicas.selfIfPresent().isFull();
}
/**
* Safely move a transfer into the live set. This must be crash-safe, and the primary invariant we need to
* preserve is a transfer is only added to the live set iff the transfer ID is present in its mutation summaries.
* <p>
* We don't validate checksums here, mostly because a transfer can be activated during a read, if one replica
* missed the TransferActivation. Transfers should not be pending for very long, and should be protected by
* internode integrity checks provided by TLS.
* <p>
* Synchronized to prevent a single activation from running multiple times if requested during read reconciliation
* and in the background via {@link ActiveLogReconciler}.
*/
public synchronized void activate(TransferActivation activation)
{
if (activated)
return;
Preconditions.checkState(isFullReplica());
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());
if (activation.isPrepare())
{
logger.info("{} Not adding SSTables to live set for dryRun {}", logPrefix(), activation);
return;
}
// Modify SSTables metadata to durably set transfer ID before importing
ImmutableCoordinatorLogOffsets logOffsets = new ImmutableCoordinatorLogOffsets.Builder()
.addTransfer(activation.transferId, sstables)
.build();
// Ensure no lingering mutation IDs, only activation IDs
for (SSTableReader sstable : sstables)
{
Preconditions.checkState(sstable.getCoordinatorLogOffsets().mutations().isEmpty());
try
{
sstable.mutateCoordinatorLogOffsetsAndReload(logOffsets);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
Preconditions.checkState(sstable.getCoordinatorLogOffsets().mutations().isEmpty());
ActivatedTransfers transfers = sstable.getCoordinatorLogOffsets().transfers();
Preconditions.checkState(!transfers.isEmpty());
}
File dst = cfs.getDirectories().getDirectoryForNewSSTables();
// Retain the original SSTables in pending/ dir on the coordinator, so future streams can get the originals, and
// we don't need to isolate activated SSTables during compaction
boolean isCoordinator = activation.transferId.hostId == ClusterMetadata.current().myNodeId().id();
logger.debug("{} {} pending SSTables for activation to {}", isCoordinator ? "Copying" : "Moving", logPrefix(), dst);
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(), isCoordinator));
// Add all SSTables atomically
cfs.getTracker().addSSTablesTracked(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().scheduleCleanup();
}
@Override
public String toString()
{
return "PendingLocalTransfer{" +
"activated=" + activated +
", range=" + range +
", 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);
}
}

8
src/java/org/apache/cassandra/replication/PullMutationsRequest.java
View File

@ -72,4 +72,12 @@ public final class PullMutationsRequest
MutationTrackingService.instance.requestMissingMutations(offsets, forHost);
}
};
@Override
public String toString()
{
return "PullMutationsRequest{" +
"offsets=" + offsets +
'}';
}
}

25
src/java/org/apache/cassandra/replication/Shard.java
View File

@ -139,7 +139,7 @@ public class Shard
{
CoordinatorLog newLog = log.withParticipants(newParticipants);
newLogs.put(newLog.logId.asLong(), newLog);
if (log == currentLocalLog)
newCurrentLocalLog = (CoordinatorLog.CoordinatorLogPrimary) newLog;
}
@ -153,9 +153,10 @@ public class Shard
MutationId nextId()
{
MutationId nextId = currentLocalLog.nextId();
if (nextId != null)
return nextId;
return maybeRotateLocalLogAndGetNextId();
if (nextId == null)
nextId = maybeRotateLocalLogAndGetNextId();
logger.trace("Issuing next MutationId {}", nextId);
return nextId;
}
// if ids overflow, we need to rotate the local log
@ -176,6 +177,17 @@ public class Shard
getOrCreate(mutationId).receivedWriteResponse(mutationId, fromHostId);
}
void finishActivation(PendingLocalTransfer transfer, TransferActivation activation)
{
getOrCreate(activation.transferId).finishActivation(transfer, activation);
}
void receivedActivationResponse(CoordinatedTransfer transfer, InetAddressAndPort onHost)
{
int onHostId = ClusterMetadata.current().directory.peerId(onHost).id();
getOrCreate(transfer.id()).receivedActivationResponse(transfer, onHostId);
}
void updateReplicatedOffsets(List<? extends Offsets> offsets, boolean durable, InetAddressAndPort onHost)
{
int onHostId = ClusterMetadata.current().directory.peerId(onHost).id();
@ -255,6 +267,11 @@ public class Shard
return replicas;
}
boolean isDurablyReconciled(ShortMutationId id)
{
return logs.get(id.logId()).isDurablyReconciled(id);
}
boolean isDurablyReconciled(long logId, CoordinatorLogOffsets<?> logOffsets)
{
return logs.get(logId).isDurablyReconciled(logOffsets);

197
src/java/org/apache/cassandra/replication/TransferActivation.java Normal file
View File

@ -0,0 +1,197 @@
/*
* 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.replication;
import java.io.IOException;
import java.util.Objects;
import com.google.common.base.Preconditions;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.NoPayload;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.TimeUUID;
/**
* Activation transitions a {@link PendingLocalTransfer} from being pending (durable on disk but not yet visible to
* reads) to live (visible to reads and compactions), by associating the streaming plan ID with a mutation ID, referred
* to as the transfer ID.
* <p>
* See {@link CoordinatedTransfer} for the lifecycle of a transfer and when a {@link TransferActivation} is sent.
*/
public class TransferActivation
{
public final TimeUUID planId;
public final ShortMutationId transferId;
public final NodeId coordinatorId;
public final Phase phase;
public enum Phase
{
PREPARE(0),
COMMIT(1);
private final int id;
private static final Phase[] ids;
static
{
ids = new Phase[values().length];
for (Phase phase : values())
ids[phase.id] = phase;
}
Phase(int id)
{
this.id = id;
}
static Phase from(int id)
{
Phase phase = ids[id];
Preconditions.checkState(phase.id == id);
return phase;
}
}
public TransferActivation(CoordinatedTransfer transfer, InetAddressAndPort peer, Phase phase)
{
this(transfer.streamResults.get(peer).planId(), transfer.id(), ClusterMetadata.current().myNodeId(), phase);
}
TransferActivation(TimeUUID planId, ShortMutationId transferId, NodeId coordinatorId, Phase phase)
{
Preconditions.checkArgument(!transferId.isNone());
Preconditions.checkNotNull(planId);
Preconditions.checkNotNull(coordinatorId);
this.planId = planId;
this.transferId = transferId;
this.coordinatorId = coordinatorId;
this.phase = phase;
}
ShortMutationId id()
{
return transferId;
}
public boolean isPrepare()
{
return phase == Phase.PREPARE;
}
public boolean isCommit()
{
return phase == Phase.COMMIT;
}
public void apply()
{
MutationTrackingService.instance.activateLocal(this);
}
public static final Serializer serializer = new Serializer();
public static class Serializer implements IVersionedSerializer<TransferActivation>
{
@Override
public void serialize(TransferActivation activate, DataOutputPlus out, int version) throws IOException
{
TimeUUID.Serializer.instance.serialize(activate.planId, out, version);
ShortMutationId.serializer.serialize(activate.id(), out, version);
NodeId.messagingSerializer.serialize(activate.coordinatorId, out, version);
out.writeByte(activate.phase.id);
}
@Override
public TransferActivation deserialize(DataInputPlus in, int version) throws IOException
{
TimeUUID planId = TimeUUID.Serializer.instance.deserialize(in, version);
ShortMutationId id = ShortMutationId.serializer.deserialize(in, version);
NodeId coordinatorId = NodeId.messagingSerializer.deserialize(in, version);
Phase phase = Phase.from(in.readByte());
return new TransferActivation(planId, id, coordinatorId, phase);
}
@Override
public long serializedSize(TransferActivation activate, int version)
{
long size = 0;
size += TimeUUID.Serializer.instance.serializedSize(activate.planId, version);
size += ShortMutationId.serializer.serializedSize(activate.id(), version);
size += NodeId.messagingSerializer.serializedSize(activate.coordinatorId, version);
size += TypeSizes.BYTE_SIZE; // Enum ordinal
return size;
}
}
public static class VerbHandler implements IVerbHandler<TransferActivation>
{
@Override
public void doVerb(Message<TransferActivation> msg) throws IOException
{
LocalTransfers.instance().executor.submit(() -> {
try
{
msg.payload.apply();
MessagingService.instance().respond(NoPayload.noPayload, msg);
}
catch (Throwable t)
{
MessagingService.instance().respondWithFailure(RequestFailureReason.forException(t), msg);
}
});
}
}
public static final VerbHandler verbHandler = new VerbHandler();
@Override
public String toString()
{
return "TransferActivation{" +
", planId=" + planId +
", transferId=" + transferId +
", coordinatorId=" + coordinatorId +
", phase=" + phase +
'}';
}
@Override
public boolean equals(Object o)
{
if (o == null || getClass() != o.getClass()) return false;
TransferActivation that = (TransferActivation) o;
return Objects.equals(planId, that.planId) && Objects.equals(transferId, that.transferId) && Objects.equals(coordinatorId, that.coordinatorId) && phase == that.phase;
}
@Override
public int hashCode()
{
return Objects.hash(planId, transferId, coordinatorId, phase);
}
}

72
src/java/org/apache/cassandra/replication/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.replication;
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 +
'}';
}
}

36
src/java/org/apache/cassandra/replication/UnknownShardException.java Normal file
View File

@ -0,0 +1,36 @@
/*
* 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.replication;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.tcm.ownership.ReplicaGroups;
public class UnknownShardException extends IllegalStateException
{
public UnknownShardException(Token token, ReplicaGroups groups)
{
super(String.format("Could not find token %s in %s", token, groups));
}
public UnknownShardException(Range<Token> range, ReplicaGroups groups)
{
super(String.format("Could not find range %s in %s", range, groups));
}
}

41
src/java/org/apache/cassandra/replication/UnreconciledMutations.java
View File

@ -27,11 +27,15 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.agrona.collections.Int2ObjectHashMap;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.TableId;
/**
@ -40,9 +44,15 @@ import org.apache.cassandra.schema.TableId;
*/
public class UnreconciledMutations
{
private static final Logger logger = LoggerFactory.getLogger(UnreconciledMutations.class);
// Mutations (single-partition)
private final Int2ObjectHashMap<Entry> statesMap = new Int2ObjectHashMap<>();
private final SortedSet<Entry> statesSet = new TreeSet<>(Entry.comparator);
// Transfers (partition-range)
private final ActivatedTransfers transfers = new ActivatedTransfers();
enum Visibility
{
PENDING, // written to the journal, but not yet to LSM
@ -135,15 +145,23 @@ public class UnreconciledMutations
public void remove(int offset)
{
Entry state = statesMap.remove(offset);
if (state != null)
if (state == null)
transfers.removeOffset(offset);
else
statesSet.remove(state);
}
public void activatedTransfer(ShortMutationId id, Collection<SSTableReader> sstables)
{
transfers.add(id, sstables);
}
public UnreconciledMutations copy()
{
UnreconciledMutations copy = new UnreconciledMutations();
copy.statesMap.putAll(statesMap);
copy.statesSet.addAll(statesSet);
copy.transfers.addAll(transfers);
return copy;
}
@ -151,12 +169,14 @@ public class UnreconciledMutations
{
Entry start = Entry.start(range.left.getToken(), range.left.kind() != PartitionPosition.Kind.MAX_BOUND);
Entry end = Entry.end(range.right.getToken(), range.right.kind() != PartitionPosition.Kind.MIN_BOUND);
transfers.forEachIntersecting(range, id -> into.add(id.offset()));
return collect(start, end, tableId, includePending, into);
}
public boolean collect(Token token, TableId tableId, boolean includePending, Offsets.OffsetReciever into)
{
SortedSet<Entry> subset = statesSet.subSet(Entry.start(token, true), Entry.end(token, true));
transfers.forEachIntersecting(token, id -> into.add(id.offset()));
return collect(subset, tableId, includePending, into);
}
@ -210,7 +230,7 @@ public class UnreconciledMutations
@VisibleForTesting
boolean equalsForTesting(UnreconciledMutations other)
{
return this.statesMap.equals(other.statesMap) && this.statesSet.equals(other.statesSet);
return this.statesMap.equals(other.statesMap) && this.statesSet.equals(other.statesSet) && this.transfers.equals(other.transfers);
}
@VisibleForTesting
@ -243,10 +263,25 @@ public class UnreconciledMutations
for (int offset = iter.start(), end = iter.end(); offset <= end; offset++)
{
ShortMutationId id = new ShortMutationId(witnessed.logId, offset);
result.addDirectly(MutationJournal.instance.read(id));
Mutation mutation = MutationJournal.instance.read(id);
if (mutation != null)
{
result.addDirectly(mutation);
continue;
}
CoordinatedTransfer transfer = LocalTransfers.instance().getActivatedTransfer(id);
if (transfer != null)
{
result.transfers.add(transfer.id(), transfer.sstables);
continue;
}
logger.error("Cannot load unknown mutation ID {}", id);
}
}
// Transfers are never present in the journal, since they're added as SSTables directly
return result;
}
}

11
src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
View File

@ -247,10 +247,19 @@ public abstract class PartialTrackedRead
augmentingOffsets.forEach(this::augment);
}
/**
* If a mutation to augment isn't present in {@link MutationJournal}, it's either a newly-activated transfer, or a
* serious bug. In the case it's a transfer, we still want to signal this to the client to retry against another
* node. This could be optimized to augment with updates from the newly-transferred SSTables.
*/
void augment(ShortMutationId mutationId)
{
Mutation mutation = MutationJournal.instance.read(mutationId);
Preconditions.checkNotNull(mutation, "Missing mutation %s", mutationId);
if (mutation == null)
{
logger.error("Could not augment read with mutation not present in journal {}", mutationId);
throw new RuntimeException(String.format("Missing mutation %s", mutationId));
}
if (!command().selectsKey(mutation.key()))
{
logger.trace("Skipping mutation {} - {} not in read range", mutationId, mutation.key());

10
src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
View File

@ -23,6 +23,9 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.agrona.collections.IntArrayList;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.gms.FailureDetector;
@ -126,6 +129,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
public void acceptMutation(TrackedRead.Id id, ShortMutationId mutationId)
{
logger.debug("Accepted mutation {} {}", id, mutationId);
Coordinator reconcile = coordinators.get(id);
if (reconcile != null && reconcile.acceptMutation(mutationId)) // could be already timed out / expired
coordinators.remove(id);
@ -155,6 +159,8 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
private static final class Coordinator
{
private static final Logger logger = LoggerFactory.getLogger(Coordinator.class);
private static final AtomicLongFieldUpdater<Coordinator> remainingUpdater =
AtomicLongFieldUpdater.newUpdater(Coordinator.class, "remaining");
private volatile long remaining; // three values packed into one atomic long
@ -343,6 +349,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
int mutations = remainingMutations(prev) + mutationsDelta;
int summaries = remainingSummaries(prev) + summariesDelta;
int syncAcks = remainingSyncAcks(prev) + syncAcksDelta;
logger.trace("[Read {}] Still waiting for {} mutations, {} summaries, {} syncAcks", id, mutations, summaries, syncAcks);
next = remaining(mutations, summaries, syncAcks);
} while (!remainingUpdater.compareAndSet(this, prev, next));
return next;
@ -369,6 +376,8 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
}
}
private static final Logger logger = LoggerFactory.getLogger(ReadReconciliations.class);
/**
* @param node node id of the remote replica from which we got the summary
* @param offsets offsets that we need to pull - from the coordinator, if alive, or from the
@ -389,6 +398,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
if (!toPull.isEmpty())
{
PullMutationsRequest pull = new PullMutationsRequest(Offsets.Immutable.copy(toPull));
logger.debug("Pulling {} from {}", pull, pullFrom);
MessagingService.instance().send(Message.out(Verb.PULL_MUTATIONS_REQ, pull), pullFrom);
}
}

28
src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.service.reads.tracked;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import com.google.common.annotations.VisibleForTesting;
@ -128,6 +129,9 @@ public class TrackedLocalReads implements ExpiredStatePurger.Expireable
// any mutations that may have arrived during initial read execution.
secondarySummary = command.createMutationSummary(true);
processDelta(read, initialSummary, secondarySummary);
// Include in summary any transfer IDs that were present for the read
secondarySummary = merge(controller.getTransferIds(), secondarySummary);
}
catch (Exception e)
{
@ -150,6 +154,30 @@ public class TrackedLocalReads implements ExpiredStatePurger.Expireable
ReadReconciliations.instance.acceptLocalSummary(readId, secondarySummary, summaryNodes);
}
private static MutationSummary merge(Iterator<ShortMutationId> transferIds, MutationSummary summary)
{
if (transferIds == null || !transferIds.hasNext())
return summary;
MutationSummary.Builder builder = new MutationSummary.Builder(summary.tableId());
// TODO: Make faster without a copy
for (int i = 0; i < summary.size(); i++)
{
MutationSummary.CoordinatorSummary coordinatorSummary = summary.get(i);
MutationSummary.CoordinatorSummary.Builder coordinatorSummaryBuilder = builder.builderForLog(coordinatorSummary.logId());
coordinatorSummaryBuilder.unreconciled.addAll(coordinatorSummary.unreconciled);
coordinatorSummaryBuilder.reconciled.addAll(coordinatorSummary.reconciled);
}
while (transferIds.hasNext())
{
ShortMutationId id = transferIds.next();
builder.builderForLog(id).unreconciled.add(id.offset());
}
return builder.build();
}
@VisibleForTesting
public static void processDelta(PartialTrackedRead read, MutationSummary initialSummary, MutationSummary secondarySummary)
{

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),
TRACKED_TRANSFER("Tracked Transfer", false, false, false);
private final String description;
private final boolean requiresViewBuild;

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

@ -73,8 +73,7 @@ public class StreamPlan
boolean connectSequentially, TimeUUID pendingRepair, PreviewKind previewKind)
{
this.streamOperation = streamOperation;
this.coordinator = new StreamCoordinator(streamOperation, connectionsPerHost, streamingFactory(),
false, connectSequentially, pendingRepair, previewKind);
this.coordinator = new StreamCoordinator(streamOperation, connectionsPerHost, streamingFactory(), false, connectSequentially, pendingRepair, previewKind);
}
/**

15
src/java/org/apache/cassandra/tcm/ownership/ReplicaGroups.java
View File

@ -49,6 +49,7 @@ import org.apache.cassandra.locator.RangesAtEndpoint;
import org.apache.cassandra.locator.RangesByEndpoint;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaCollection;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.serialization.PartitionerAwareMetadataSerializer;
@ -123,6 +124,20 @@ public class ReplicaGroups
return null;
}
/**
* Intended to be used for lookup of range-shards, such as those managed by {@link MutationTrackingService}.
*/
public VersionedEndpoints.ForRange matchToken(Token token)
{
int pos = ordering.binarySearchAsymmetric(ranges, token, AsymmetricOrdering.Op.CEIL);
if (pos >= 0 && pos < ranges.size() && ranges.get(pos).contains(token))
return endpoints.get(pos);
// Last range can wrap around
if (!ranges.isEmpty() && pos == ranges.size() && ranges.get(pos - 1).contains(token))
return endpoints.get(pos - 1);
return null;
}
/**
* This method is intended to be used on read/write path, not forRange.
*/

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-CompressionInfo.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Digest.crc32
View File

@ -1 +1 @@
2530067741
2415910404

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Rows.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Statistics.db
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-TOC.txt
View File

@ -1,8 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Data.db
Digest.crc32
Filter.db
Partitions.db
Rows.db
Statistics.db
TOC.txt

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-CompressionInfo.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Digest.crc32
View File

@ -1 +1 @@
1370392555
2922608909

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Rows.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Statistics.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-CompressionInfo.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Digest.crc32
View File

@ -1 +1 @@
53972413
3585813939

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Rows.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Statistics.db
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-TOC.txt
View File

@ -1,8 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Data.db
Digest.crc32
Filter.db
Partitions.db
Rows.db
Statistics.db
TOC.txt

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Digest.crc32
View File

@ -1 +1 @@
3597571582
851105930

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Statistics.db
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-TOC.txt
View File

@ -1,8 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Data.db
Digest.crc32
Filter.db
Partitions.db
Rows.db
Statistics.db
TOC.txt

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Digest.crc32
View File

@ -1 +1 @@
3514184066
4193692618

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Statistics.db
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-TOC.txt
View File

@ -1,8 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Data.db
Digest.crc32
Filter.db
Partitions.db
Rows.db
Statistics.db
TOC.txt

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Digest.crc32
View File

@ -1 +1 @@
1960712659
2761811502

BIN
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Statistics.db
View File

Binary file not shown.

8
test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-TOC.txt
View File

@ -1,8 +1,8 @@
Data.db
Statistics.db
Digest.crc32
TOC.txt
CompressionInfo.db
Data.db
Digest.crc32
Filter.db
Partitions.db
Rows.db
Statistics.db
TOC.txt

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-CompressionInfo.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Digest.crc32
View File

@ -1 +1 @@
2672104528
733768223

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Index.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Statistics.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-CompressionInfo.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Digest.crc32
View File

@ -1 +1 @@
2155890726
1457015109

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Index.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Statistics.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-CompressionInfo.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Digest.crc32
View File

@ -1 +1 @@
1361141780
880796784

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Index.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Statistics.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Digest.crc32
View File

@ -1 +1 @@
1375280580
2614540741

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Statistics.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Digest.crc32
View File

@ -1 +1 @@
575257059
19151750

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Statistics.db
View File

Binary file not shown.

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Data.db
View File

Binary file not shown.

2
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Digest.crc32
View File

@ -1 +1 @@
383358291
4267206168

BIN
test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Statistics.db
View File

Binary file not shown.

10
test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
View File

@ -1763,4 +1763,14 @@ public class ClusterUtils
.collect(Collectors.toList());
});
}
public static Integer instanceId(IInvokableInstance instance)
{
return instance.callOnInstance(() -> {
ClassLoader cl = Thread.currentThread().getContextClassLoader();
if (cl instanceof InstanceClassLoader)
return ((InstanceClassLoader) cl).getInstanceId();
return null;
});
}
}

Some files were not shown because too many files have changed in this diff Show More