You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2026/04/14 21:47:13 UTC
(cassandra) 24/36: CEP-45: Bulk transfer
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 407e19021bd7af56da00c140ae7274440872d24d
Author: Abe Ratnofsky <ab...@aber.io>
AuthorDate: Tue Sep 2 13:07:38 2025 -0400
CEP-45: Bulk transfer
patch by Abe Ratnofsky; reviewed by Caleb Rackliffe and Blake Eggleston for CASSANDRA-20383
---
.../org/apache/cassandra/db/ColumnFamilyStore.java | 36 +-
src/java/org/apache/cassandra/db/Directories.java | 43 +-
.../cassandra/db/PartitionRangeReadCommand.java | 2 +
.../cassandra/db/ReadExecutionController.java | 48 ++
.../org/apache/cassandra/db/SSTableImporter.java | 30 +-
.../cassandra/db/SinglePartitionReadCommand.java | 7 +
.../cassandra/db/compaction/CompactionManager.java | 2 +-
.../cassandra/db/compaction/CompactionTask.java | 2 +
.../org/apache/cassandra/db/lifecycle/Tracker.java | 24 +
.../org/apache/cassandra/db/lifecycle/View.java | 2 +-
.../CassandraEntireSSTableStreamReader.java | 9 +-
.../db/streaming/CassandraStreamReader.java | 13 +-
.../db/streaming/CassandraStreamReceiver.java | 60 +-
.../io/sstable/RangeAwarePendingSSTableWriter.java | 45 +
.../io/sstable/RangeAwareSSTableWriter.java | 14 +-
.../cassandra/io/sstable/format/SSTableReader.java | 12 +
.../io/sstable/metadata/IMetadataSerializer.java | 6 +
.../io/sstable/metadata/MetadataSerializer.java | 10 +
.../io/sstable/metadata/StatsMetadata.java | 29 +
src/java/org/apache/cassandra/net/Verb.java | 7 +
.../cassandra/replication/ActivatedTransfers.java | 287 +++++++
.../cassandra/replication/ActiveLogReconciler.java | 81 +-
.../cassandra/replication/CoordinatedTransfer.java | 720 ++++++++++++++++
.../replication/CoordinatedTransfers.java | 90 ++
.../cassandra/replication/CoordinatorLog.java | 130 ++-
.../replication/CoordinatorLogOffsets.java | 29 +-
.../cassandra/replication/ForwardedWrite.java | 2 +-
.../ImmutableCoordinatorLogOffsets.java | 257 ++++--
.../cassandra/replication/LocalTransfers.java | 290 +++++++
.../replication/MutableCoordinatorLogOffsets.java | 6 +-
.../replication/MutationTrackingService.java | 129 ++-
.../cassandra/replication/Node2OffsetsMap.java | 8 +
.../NonBlockingCoordinatorLogOffsets.java | 33 +-
.../replication/PendingLocalTransfer.java | 243 ++++++
.../replication/PullMutationsRequest.java | 8 +
.../org/apache/cassandra/replication/Shard.java | 25 +-
.../cassandra/replication/TransferActivation.java | 197 +++++
.../cassandra/replication/TransferFailed.java | 72 ++
...rLogOffsets.java => UnknownShardException.java} | 27 +-
.../replication/UnreconciledMutations.java | 41 +-
.../service/reads/tracked/PartialTrackedRead.java | 11 +-
.../service/reads/tracked/ReadReconciliations.java | 10 +
.../service/reads/tracked/TrackedLocalReads.java | 28 +
.../cassandra/streaming/StreamOperation.java | 3 +-
.../org/apache/cassandra/streaming/StreamPlan.java | 3 +-
.../cassandra/tcm/ownership/ReplicaGroups.java | 15 +
.../legacy_db_clust/db-51-bti-CompressionInfo.db | Bin 207 -> 207 bytes
.../legacy_db_clust/db-51-bti-Data.db | Bin 8496 -> 8662 bytes
.../legacy_db_clust/db-51-bti-Digest.crc32 | 2 +-
.../legacy_db_clust/db-51-bti-Rows.db | Bin 508 -> 508 bytes
.../legacy_db_clust/db-51-bti-Statistics.db | Bin 7313 -> 7334 bytes
.../legacy_db_clust/db-51-bti-TOC.txt | 6 +-
.../db-51-bti-CompressionInfo.db | Bin 207 -> 207 bytes
.../db-51-bti-Data.db | Bin 8451 -> 8659 bytes
.../db-51-bti-Digest.crc32 | 2 +-
.../db-51-bti-Rows.db | Bin 508 -> 508 bytes
.../db-51-bti-Statistics.db | Bin 7333 -> 7334 bytes
.../db-51-bti-CompressionInfo.db | Bin 199 -> 199 bytes
.../legacy_db_clust_counter/db-51-bti-Data.db | Bin 7332 -> 8290 bytes
.../legacy_db_clust_counter/db-51-bti-Digest.crc32 | 2 +-
.../legacy_db_clust_counter/db-51-bti-Rows.db | Bin 508 -> 508 bytes
.../db-51-bti-Statistics.db | Bin 7322 -> 7343 bytes
.../legacy_db_clust_counter/db-51-bti-TOC.txt | 6 +-
.../legacy_db_simple/db-51-bti-Data.db | Bin 78 -> 80 bytes
.../legacy_db_simple/db-51-bti-Digest.crc32 | 2 +-
.../legacy_db_simple/db-51-bti-Statistics.db | Bin 4823 -> 4844 bytes
.../legacy_db_simple/db-51-bti-TOC.txt | 6 +-
.../legacy_db_simple_counter/db-51-bti-Data.db | Bin 132 -> 141 bytes
.../db-51-bti-Digest.crc32 | 2 +-
.../db-51-bti-Statistics.db | Bin 4832 -> 4853 bytes
.../legacy_db_simple_counter/db-51-bti-TOC.txt | 6 +-
.../legacy_db_tuple/db-51-bti-Data.db | Bin 300 -> 301 bytes
.../legacy_db_tuple/db-51-bti-Digest.crc32 | 2 +-
.../legacy_db_tuple/db-51-bti-Statistics.db | Bin 5802 -> 5807 bytes
.../legacy_db_tuple/db-51-bti-TOC.txt | 6 +-
.../legacy_ob_clust/ob-51-big-CompressionInfo.db | Bin 207 -> 207 bytes
.../legacy_ob_clust/ob-51-big-Data.db | Bin 8568 -> 8653 bytes
.../legacy_ob_clust/ob-51-big-Digest.crc32 | 2 +-
.../legacy_ob_clust/ob-51-big-Index.db | Bin 157498 -> 157498 bytes
.../legacy_ob_clust/ob-51-big-Statistics.db | Bin 7333 -> 7334 bytes
.../ob-51-big-CompressionInfo.db | Bin 207 -> 207 bytes
.../ob-51-big-Data.db | Bin 8623 -> 8650 bytes
.../ob-51-big-Digest.crc32 | 2 +-
.../ob-51-big-Index.db | Bin 157498 -> 157498 bytes
.../ob-51-big-Statistics.db | Bin 7333 -> 7334 bytes
.../ob-51-big-CompressionInfo.db | Bin 199 -> 199 bytes
.../legacy_ob_clust_counter/ob-51-big-Data.db | Bin 7909 -> 8260 bytes
.../legacy_ob_clust_counter/ob-51-big-Digest.crc32 | 2 +-
.../legacy_ob_clust_counter/ob-51-big-Index.db | Bin 157498 -> 157498 bytes
.../ob-51-big-Statistics.db | Bin 7342 -> 7343 bytes
.../legacy_ob_simple/ob-51-big-Data.db | Bin 79 -> 80 bytes
.../legacy_ob_simple/ob-51-big-Digest.crc32 | 2 +-
.../legacy_ob_simple/ob-51-big-Statistics.db | Bin 4843 -> 4844 bytes
.../legacy_ob_simple_counter/ob-51-big-Data.db | Bin 139 -> 141 bytes
.../ob-51-big-Digest.crc32 | 2 +-
.../ob-51-big-Statistics.db | Bin 4852 -> 4853 bytes
.../legacy_ob_tuple/ob-51-big-Data.db | Bin 301 -> 303 bytes
.../legacy_ob_tuple/ob-51-big-Digest.crc32 | 2 +-
.../legacy_ob_tuple/ob-51-big-Statistics.db | Bin 5806 -> 5807 bytes
.../cassandra/distributed/shared/ClusterUtils.java | 10 +
.../test/tracking/BulkTransfersTest.java | 949 +++++++++++++++++++++
.../unit/org/apache/cassandra/ServerTestUtils.java | 4 +
.../db/CoordinatorLogOffsetsLifecycleTest.java | 27 +-
.../db/compaction/AntiCompactionBytemanTest.java | 2 +-
.../apache/cassandra/db/lifecycle/TrackerTest.java | 2 +-
.../replication/ActivatedTransfersTest.java | 280 ++++++
.../replication/CoordinatorLogOffsetsTest.java | 58 +-
.../cassandra/replication/LocalTransfersTest.java | 344 ++++++++
.../TransferActivationSerializationTest.java | 75 ++
.../cassandra/tcm/ownership/ReplicaGroupsTest.java | 228 +++++
.../cassandra/utils/CassandraGenerators.java | 45 +
111 files changed, 4984 insertions(+), 240 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index d7488a36d6..793918c545 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -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;
diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java
index 236b7186b2..f64e741896 100644
--- a/src/java/org/apache/cassandra/db/Directories.java
+++ b/src/java/org/apache/cassandra/db/Directories.java
@@ -74,6 +74,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.snapshot.SnapshotManifest;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
@@ -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();
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index 962065c3a6..f3d6b01eb5 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -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.
diff --git a/src/java/org/apache/cassandra/db/ReadExecutionController.java b/src/java/org/apache/cassandra/db/ReadExecutionController.java
index 85b17d4690..513199fdd5 100644
--- a/src/java/org/apache/cassandra/db/ReadExecutionController.java
+++ b/src/java/org/apache/cassandra/db/ReadExecutionController.java
@@ -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();
+ }
}
diff --git a/src/java/org/apache/cassandra/db/SSTableImporter.java b/src/java/org/apache/cassandra/db/SSTableImporter.java
index 5931f39418..d6b4fe79bf 100644
--- a/src/java/org/apache/cassandra/db/SSTableImporter.java
+++ b/src/java/org/apache/cassandra/db/SSTableImporter.java
@@ -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
*
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index ed2fbdc645..4563245d6f 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@ -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)
{
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index d322f0c6f2..87fd050d06 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -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(),
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index a657552b07..d052c7d3d8 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -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();
}
diff --git a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
index b207d86165..a464613f7c 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
@@ -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);
}
diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java
index fe188e37c3..1d373203e7 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/View.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/View.java
@@ -403,4 +403,4 @@ public class View
}
};
}
-}
\ No newline at end of file
+}
diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
index 1f178a482b..535c5ff7de 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
@@ -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));
diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java
index d24901252b..e400774e4c 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java
@@ -47,6 +47,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.UnknownColumnException;
+import org.apache.cassandra.io.sstable.RangeAwarePendingSSTableWriter;
import org.apache.cassandra.io.sstable.RangeAwareSSTableWriter;
import org.apache.cassandra.io.sstable.SSTableMultiWriter;
import org.apache.cassandra.io.sstable.SSTableSimpleIterator;
@@ -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);
}
diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
index c448d5edbe..cef0794209 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
@@ -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);
}
}
}
diff --git a/src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java
new file mode 100644
index 0000000000..232cf0af62
--- /dev/null
+++ b/src/java/org/apache/cassandra/io/sstable/RangeAwarePendingSSTableWriter.java
@@ -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);
+ }
+}
diff --git a/src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java
index e30f533b26..a4c15ff53e 100644
--- a/src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/RangeAwareSSTableWriter.java
@@ -32,6 +32,7 @@ import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.util.File;
import org.apache.cassandra.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());
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index 0cace7218a..7262af49cd 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@ -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>
diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/IMetadataSerializer.java b/src/java/org/apache/cassandra/io/sstable/metadata/IMetadataSerializer.java
index e8299466c7..1b99222bde 100644
--- a/src/java/org/apache/cassandra/io/sstable/metadata/IMetadataSerializer.java
+++ b/src/java/org/apache/cassandra/io/sstable/metadata/IMetadataSerializer.java
@@ -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.
*/
diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/MetadataSerializer.java b/src/java/org/apache/cassandra/io/sstable/metadata/MetadataSerializer.java
index e253d3d87b..36325eea5f 100644
--- a/src/java/org/apache/cassandra/io/sstable/metadata/MetadataSerializer.java
+++ b/src/java/org/apache/cassandra/io/sstable/metadata/MetadataSerializer.java
@@ -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));
diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
index ca6c66aa43..21d5dedeb3 100644
--- a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
+++ b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
@@ -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)
{
diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java
index 79362bf5dc..f9d3b2e447 100644
--- a/src/java/org/apache/cassandra/net/Verb.java
+++ b/src/java/org/apache/cassandra/net/Verb.java
@@ -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 ),
diff --git a/src/java/org/apache/cassandra/replication/ActivatedTransfers.java b/src/java/org/apache/cassandra/replication/ActivatedTransfers.java
new file mode 100644
index 0000000000..bcee80ecba
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/ActivatedTransfers.java
@@ -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);
+ }
+ };
+}
diff --git a/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java b/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
index 54ded3766c..8667c76601 100644
--- a/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
+++ b/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
@@ -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;
diff --git a/src/java/org/apache/cassandra/replication/CoordinatedTransfer.java b/src/java/org/apache/cassandra/replication/CoordinatedTransfer.java
new file mode 100644
index 0000000000..d82b27c9b2
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/CoordinatedTransfer.java
@@ -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 +
+ '}';
+ }
+}
diff --git a/src/java/org/apache/cassandra/replication/CoordinatedTransfers.java b/src/java/org/apache/cassandra/replication/CoordinatedTransfers.java
new file mode 100644
index 0000000000..2adb546fc5
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/CoordinatedTransfers.java
@@ -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 + '}';
+ }
+}
diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLog.java b/src/java/org/apache/cassandra/replication/CoordinatorLog.java
index 6f4a4aed4c..9dac9af12a 100644
--- a/src/java/org/apache/cassandra/replication/CoordinatorLog.java
+++ b/src/java/org/apache/cassandra/replication/CoordinatorLog.java
@@ -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)
{
diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java
index dfe34f5800..5a394fc08d 100644
--- a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java
+++ b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java
@@ -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();
}
diff --git a/src/java/org/apache/cassandra/replication/ForwardedWrite.java b/src/java/org/apache/cassandra/replication/ForwardedWrite.java
index c6014257e8..e5e2bf8d8b 100644
--- a/src/java/org/apache/cassandra/replication/ForwardedWrite.java
+++ b/src/java/org/apache/cassandra/replication/ForwardedWrite.java
@@ -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.
diff --git a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
index e4115d9690..af10da0073 100644
--- a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
+++ b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
@@ -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)
- {
- Offsets.Immutable offsets = ids.get(logId);
- if (offsets == null)
- return new Offsets.Immutable(new CoordinatorLogId(logId));
- return offsets;
- }
+ private final ImmutableMutations mutations;
+ private final ActivatedTransfers transfers;
- @Override
- public int size()
+ private ImmutableCoordinatorLogOffsets(Builder builder)
{
- return ids.size();
+ // 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);
}
- public boolean isEmpty()
+ @Override
+ public Mutations<Offsets.Immutable> mutations()
{
- 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)
- {
- // 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)
+ @Override
+ public String toString()
{
- 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;
+ }
+ };
+ }
}
diff --git a/src/java/org/apache/cassandra/replication/LocalTransfers.java b/src/java/org/apache/cassandra/replication/LocalTransfers.java
new file mode 100644
index 0000000000..876bbeb47e
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/LocalTransfers.java
@@ -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);
+ };
+}
diff --git a/src/java/org/apache/cassandra/replication/MutableCoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/MutableCoordinatorLogOffsets.java
index e47e6245d9..9ecb389554 100644
--- a/src/java/org/apache/cassandra/replication/MutableCoordinatorLogOffsets.java
+++ b/src/java/org/apache/cassandra/replication/MutableCoordinatorLogOffsets.java
@@ -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);
}
}
diff --git a/src/java/org/apache/cassandra/replication/MutationTrackingService.java b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
index 4009dbd39a..5a65ca7d47 100644
--- a/src/java/org/apache/cassandra/replication/MutationTrackingService.java
+++ b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
@@ -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()
diff --git a/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java b/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java
index 1001d855cd..ac6fcc0daf 100644
--- a/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java
+++ b/src/java/org/apache/cassandra/replication/Node2OffsetsMap.java
@@ -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 +
+ '}';
+ }
}
diff --git a/src/java/org/apache/cassandra/replication/NonBlockingCoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/NonBlockingCoordinatorLogOffsets.java
index 857d6fc999..3f54ad847b 100644
--- a/src/java/org/apache/cassandra/replication/NonBlockingCoordinatorLogOffsets.java
+++ b/src/java/org/apache/cassandra/replication/NonBlockingCoordinatorLogOffsets.java
@@ -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>
diff --git a/src/java/org/apache/cassandra/replication/PendingLocalTransfer.java b/src/java/org/apache/cassandra/replication/PendingLocalTransfer.java
new file mode 100644
index 0000000000..a2c0d833a7
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/PendingLocalTransfer.java
@@ -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);
+ }
+}
diff --git a/src/java/org/apache/cassandra/replication/PullMutationsRequest.java b/src/java/org/apache/cassandra/replication/PullMutationsRequest.java
index ec5632ab5b..d8e6f0febb 100644
--- a/src/java/org/apache/cassandra/replication/PullMutationsRequest.java
+++ b/src/java/org/apache/cassandra/replication/PullMutationsRequest.java
@@ -72,4 +72,12 @@ public final class PullMutationsRequest
MutationTrackingService.instance.requestMissingMutations(offsets, forHost);
}
};
+
+ @Override
+ public String toString()
+ {
+ return "PullMutationsRequest{" +
+ "offsets=" + offsets +
+ '}';
+ }
}
diff --git a/src/java/org/apache/cassandra/replication/Shard.java b/src/java/org/apache/cassandra/replication/Shard.java
index 7274127fcb..924ae7f357 100644
--- a/src/java/org/apache/cassandra/replication/Shard.java
+++ b/src/java/org/apache/cassandra/replication/Shard.java
@@ -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);
diff --git a/src/java/org/apache/cassandra/replication/TransferActivation.java b/src/java/org/apache/cassandra/replication/TransferActivation.java
new file mode 100644
index 0000000000..ba0c918286
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/TransferActivation.java
@@ -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);
+ }
+}
diff --git a/src/java/org/apache/cassandra/replication/TransferFailed.java b/src/java/org/apache/cassandra/replication/TransferFailed.java
new file mode 100644
index 0000000000..6a956df870
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/TransferFailed.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.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 +
+ '}';
+ }
+}
diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/UnknownShardException.java
similarity index 54%
copy from src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java
copy to src/java/org/apache/cassandra/replication/UnknownShardException.java
index dfe34f5800..b5c9bb25ab 100644
--- a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java
+++ b/src/java/org/apache/cassandra/replication/UnknownShardException.java
@@ -18,22 +18,19 @@
package org.apache.cassandra.replication;
-import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.tcm.ownership.ReplicaGroups;
-/**
- * Mutation ID offsets present in this SSTable for each coordinator log, to determine whether an SSTable is reconciled
- * or not.
- * <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 class UnknownShardException extends IllegalStateException
{
- O offsets(long logId);
- int size();
+ public UnknownShardException(Token token, ReplicaGroups groups)
+ {
+ super(String.format("Could not find token %s in %s", token, groups));
+ }
- ImmutableCoordinatorLogOffsets NONE = new ImmutableCoordinatorLogOffsets.Builder(0).build();
+ public UnknownShardException(Range<Token> range, ReplicaGroups groups)
+ {
+ super(String.format("Could not find range %s in %s", range, groups));
+ }
}
diff --git a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java
index 036f624b42..314e9303d7 100644
--- a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java
+++ b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java
@@ -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;
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
index 11cb4cdffc..cebac78909 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
@@ -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());
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
index 13e1e69d10..33d3c68e2f 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
@@ -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);
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java b/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
index aa58dc3319..9994989913 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
@@ -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)
{
diff --git a/src/java/org/apache/cassandra/streaming/StreamOperation.java b/src/java/org/apache/cassandra/streaming/StreamOperation.java
index b1c5908f7f..45dd7e9a2c 100644
--- a/src/java/org/apache/cassandra/streaming/StreamOperation.java
+++ b/src/java/org/apache/cassandra/streaming/StreamOperation.java
@@ -26,7 +26,8 @@ public enum StreamOperation
BOOTSTRAP("Bootstrap", false, true, false),
REBUILD("Rebuild", false, true, false),
BULK_LOAD("Bulk Load", true, false, false),
- REPAIR("Repair", true, false, true);
+ REPAIR("Repair", true, false, true),
+ TRACKED_TRANSFER("Tracked Transfer", false, false, false);
private final String description;
private final boolean requiresViewBuild;
diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java
index 7fa7ce82bd..631de61612 100644
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@ -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);
}
/**
diff --git a/src/java/org/apache/cassandra/tcm/ownership/ReplicaGroups.java b/src/java/org/apache/cassandra/tcm/ownership/ReplicaGroups.java
index 9cffdde9e2..e496840baa 100644
--- a/src/java/org/apache/cassandra/tcm/ownership/ReplicaGroups.java
+++ b/src/java/org/apache/cassandra/tcm/ownership/ReplicaGroups.java
@@ -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.
*/
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-CompressionInfo.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-CompressionInfo.db
index ba7ce3f01a..2591570214 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-CompressionInfo.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-CompressionInfo.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Data.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Data.db
index ac900e5814..390767c775 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Data.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Data.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Digest.crc32 b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Digest.crc32
index 048ad49aaf..c90437f307 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Digest.crc32
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Digest.crc32
@@ -1 +1 @@
-2530067741
\ No newline at end of file
+2415910404
\ No newline at end of file
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Rows.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Rows.db
index b1ffb7400f..6d0a448d65 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Rows.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Rows.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Statistics.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Statistics.db
index 8a8a4f1f4a..c213386bcf 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Statistics.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-Statistics.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-TOC.txt b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-TOC.txt
index 298910cfdc..5bfa06ac54 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-TOC.txt
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust/db-51-bti-TOC.txt
@@ -1,8 +1,8 @@
+CompressionInfo.db
Data.db
-Statistics.db
Digest.crc32
-TOC.txt
-CompressionInfo.db
Filter.db
Partitions.db
Rows.db
+Statistics.db
+TOC.txt
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-CompressionInfo.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-CompressionInfo.db
index 4528bde73f..389b20f1e1 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-CompressionInfo.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-CompressionInfo.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Data.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Data.db
index 55c580d0b9..28d8bda9fa 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Data.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Data.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Digest.crc32 b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Digest.crc32
index 024f42fedf..bd884f1aec 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Digest.crc32
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Digest.crc32
@@ -1 +1 @@
-1370392555
\ No newline at end of file
+2922608909
\ No newline at end of file
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Rows.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Rows.db
index b923a40bac..6d0a448d65 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Rows.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Rows.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Statistics.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Statistics.db
index 4311795c8b..ec17d2798b 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Statistics.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_be_index_summary/db-51-bti-Statistics.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-CompressionInfo.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-CompressionInfo.db
index 867a774cb2..f878d2fbd9 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-CompressionInfo.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-CompressionInfo.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Data.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Data.db
index 3f530d2cb6..554b8ce1df 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Data.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Data.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Digest.crc32 b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Digest.crc32
index ac6e4ec438..89d8ecfd51 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Digest.crc32
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Digest.crc32
@@ -1 +1 @@
-53972413
\ No newline at end of file
+3585813939
\ No newline at end of file
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Rows.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Rows.db
index 43d4662852..e6027e61f8 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Rows.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Rows.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Statistics.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Statistics.db
index 1c7f09c9d7..15b371e8ab 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Statistics.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-Statistics.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-TOC.txt b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-TOC.txt
index 298910cfdc..5bfa06ac54 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-TOC.txt
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_clust_counter/db-51-bti-TOC.txt
@@ -1,8 +1,8 @@
+CompressionInfo.db
Data.db
-Statistics.db
Digest.crc32
-TOC.txt
-CompressionInfo.db
Filter.db
Partitions.db
Rows.db
+Statistics.db
+TOC.txt
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Data.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Data.db
index f082ef1be0..7ccb6bc5b2 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Data.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Data.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Digest.crc32 b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Digest.crc32
index fd953e56ba..6979c2eeee 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Digest.crc32
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Digest.crc32
@@ -1 +1 @@
-3597571582
\ No newline at end of file
+851105930
\ No newline at end of file
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Statistics.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Statistics.db
index 3bc02aa78a..b1e5e43221 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Statistics.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-Statistics.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-TOC.txt b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-TOC.txt
index 298910cfdc..5bfa06ac54 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-TOC.txt
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple/db-51-bti-TOC.txt
@@ -1,8 +1,8 @@
+CompressionInfo.db
Data.db
-Statistics.db
Digest.crc32
-TOC.txt
-CompressionInfo.db
Filter.db
Partitions.db
Rows.db
+Statistics.db
+TOC.txt
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Data.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Data.db
index 6bf3d6ff53..c743838b5d 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Data.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Data.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Digest.crc32 b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Digest.crc32
index 687d787888..7fa37f31b3 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Digest.crc32
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Digest.crc32
@@ -1 +1 @@
-3514184066
\ No newline at end of file
+4193692618
\ No newline at end of file
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Statistics.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Statistics.db
index e8b44ee599..0eeb754e2f 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Statistics.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-Statistics.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-TOC.txt b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-TOC.txt
index 298910cfdc..5bfa06ac54 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-TOC.txt
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_simple_counter/db-51-bti-TOC.txt
@@ -1,8 +1,8 @@
+CompressionInfo.db
Data.db
-Statistics.db
Digest.crc32
-TOC.txt
-CompressionInfo.db
Filter.db
Partitions.db
Rows.db
+Statistics.db
+TOC.txt
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Data.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Data.db
index aa261ff13b..3aa1ac1839 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Data.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Data.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Digest.crc32 b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Digest.crc32
index 0c820c94b7..4d83fd3453 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Digest.crc32
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Digest.crc32
@@ -1 +1 @@
-1960712659
\ No newline at end of file
+2761811502
\ No newline at end of file
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Statistics.db b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Statistics.db
index a3c99f13c9..6c3cfb2141 100644
Binary files a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Statistics.db and b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-Statistics.db differ
diff --git a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-TOC.txt b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-TOC.txt
index 298910cfdc..5bfa06ac54 100644
--- a/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-TOC.txt
+++ b/test/data/legacy-sstables/db/legacy_tables/legacy_db_tuple/db-51-bti-TOC.txt
@@ -1,8 +1,8 @@
+CompressionInfo.db
Data.db
-Statistics.db
Digest.crc32
-TOC.txt
-CompressionInfo.db
Filter.db
Partitions.db
Rows.db
+Statistics.db
+TOC.txt
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-CompressionInfo.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-CompressionInfo.db
index efba218e45..6a4bc503e6 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-CompressionInfo.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-CompressionInfo.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Data.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Data.db
index fbc8ce33b2..40356af602 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Data.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Data.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Digest.crc32 b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Digest.crc32
index f0bd564de9..e38e8e8b84 100644
--- a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Digest.crc32
+++ b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Digest.crc32
@@ -1 +1 @@
-2672104528
\ No newline at end of file
+733768223
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Index.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Index.db
index de64643dd9..c87b8775e3 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Index.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Index.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Statistics.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Statistics.db
index ae9b9467a7..95f78c9d22 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Statistics.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust/ob-51-big-Statistics.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-CompressionInfo.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-CompressionInfo.db
index 962fd24580..2b305a35fa 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-CompressionInfo.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-CompressionInfo.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Data.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Data.db
index 75270dc694..58a6fae174 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Data.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Data.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Digest.crc32 b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Digest.crc32
index eb2c8d0807..a047e36a2b 100644
--- a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Digest.crc32
+++ b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Digest.crc32
@@ -1 +1 @@
-2155890726
\ No newline at end of file
+1457015109
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Index.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Index.db
index bd25a71d86..c87b8775e3 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Index.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Index.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Statistics.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Statistics.db
index 1d4529efcc..9840fc1f46 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Statistics.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_be_index_summary/ob-51-big-Statistics.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-CompressionInfo.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-CompressionInfo.db
index e661f52267..13ceb02c96 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-CompressionInfo.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-CompressionInfo.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Data.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Data.db
index 1912950610..fadf92bb8b 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Data.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Data.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Digest.crc32 b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Digest.crc32
index a570cc66c8..807e80dc37 100644
--- a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Digest.crc32
+++ b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Digest.crc32
@@ -1 +1 @@
-1361141780
\ No newline at end of file
+880796784
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Index.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Index.db
index 4d6412ce59..e0f9d44d9b 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Index.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Index.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Statistics.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Statistics.db
index 7398ca55ce..99f23ae3b9 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Statistics.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_clust_counter/ob-51-big-Statistics.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Data.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Data.db
index 0e5836f0a9..30c195a10a 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Data.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Data.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Digest.crc32 b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Digest.crc32
index a83276d891..165812c255 100644
--- a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Digest.crc32
+++ b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Digest.crc32
@@ -1 +1 @@
-1375280580
\ No newline at end of file
+2614540741
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Statistics.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Statistics.db
index 1c84c471c1..3b1520a649 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Statistics.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple/ob-51-big-Statistics.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Data.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Data.db
index 060f1a3fa2..611e158244 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Data.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Data.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Digest.crc32 b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Digest.crc32
index a880c21aeb..0e892223f0 100644
--- a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Digest.crc32
+++ b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Digest.crc32
@@ -1 +1 @@
-575257059
\ No newline at end of file
+19151750
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Statistics.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Statistics.db
index 23503f1c02..3e58e0c3b2 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Statistics.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_simple_counter/ob-51-big-Statistics.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Data.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Data.db
index 40a42c6bd3..d8e9eab6f2 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Data.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Data.db differ
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Digest.crc32 b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Digest.crc32
index a38308e51f..c52c5f658f 100644
--- a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Digest.crc32
+++ b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Digest.crc32
@@ -1 +1 @@
-383358291
\ No newline at end of file
+4267206168
\ No newline at end of file
diff --git a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Statistics.db b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Statistics.db
index 163fede284..c5de8b8565 100644
Binary files a/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Statistics.db and b/test/data/legacy-sstables/ob/legacy_tables/legacy_ob_tuple/ob-51-big-Statistics.db differ
diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
index 19986671ec..41cb288459 100644
--- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
+++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
@@ -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;
+ });
+ }
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/BulkTransfersTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/BulkTransfersTest.java
new file mode 100644
index 0000000000..2c85dd81b9
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/BulkTransfersTest.java
@@ -0,0 +1,949 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.tracking;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import net.bytebuddy.implementation.bind.annotation.This;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.streaming.CassandraStreamReceiver;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInstanceInitializer;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.shared.Uninterruptibles;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.test.sai.SAIUtil;
+import org.apache.cassandra.exceptions.ReadTimeoutException;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.sstable.CQLSSTableWriter;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.replication.ActivatedTransfers;
+import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
+import org.apache.cassandra.replication.MutationSummary;
+import org.apache.cassandra.replication.MutationTrackingService;
+import org.apache.cassandra.replication.ShortMutationId;
+import org.apache.cassandra.replication.TransferActivation;
+import org.apache.cassandra.replication.UnknownShardException;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ownership.DataPlacement;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.assertj.core.api.Assertions;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static net.bytebuddy.matcher.ElementMatchers.takesNoArguments;
+import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
+import static org.apache.cassandra.distributed.shared.AssertUtils.row;
+import static org.apache.cassandra.replication.TransferActivation.Phase.COMMIT;
+import static org.apache.cassandra.replication.TransferActivation.Phase.PREPARE;
+
+/**
+ * For now, tracked import with a replica down is not supported. The intention is to support this scenario by allowing
+ * users to provide a {@link ConsistencyLevel} for tracked import operations, where the import will complete if
+ * sufficient replicas acknowledge the transfer and activate it.
+ */
+public class BulkTransfersTest extends TestBaseImpl
+{
+ private static final Logger logger = LoggerFactory.getLogger(BulkTransfersTest.class);
+
+ private static final String TABLE = "tbl";
+ private static final String KEYSPACE_TABLE = String.format("%s.%s", KEYSPACE, TABLE);
+ private static final String TABLE_SCHEMA_CQL = String.format(withKeyspace("CREATE TABLE %s." + TABLE + " (k int primary key, v int);"));
+
+ private static final int IMPORT_PK = 1;
+ private static final Token IMPORT_TOKEN = Murmur3Partitioner.instance.getToken(Int32Type.instance.decompose(IMPORT_PK));
+ private static final int NODES = 3;
+
+ private static final IIsolatedExecutor.SerializableConsumer<SSTableReader> TRANSFERS_EXIST = sstable -> {
+ Assertions.assertThat(sstable.getCoordinatorLogOffsets().transfers())
+ .isNotEmpty();
+ Assertions.assertThat(sstable.isRepaired()).isFalse();
+ };
+ private static final IIsolatedExecutor.SerializableConsumer<SSTableReader> TRANSFERS_EMPTY = sstable -> {
+ Assertions.assertThat(sstable.getCoordinatorLogOffsets().transfers())
+ .isEmpty();
+ Assertions.assertThat(sstable.isRepaired()).isTrue();
+ };
+ private static final IIsolatedExecutor.SerializableConsumer<SSTableReader> NOOP = sstable -> {};
+
+ @Test
+ public void importHappyPath() throws Throwable
+ {
+ try (Cluster cluster = cluster())
+ {
+ createSchema(cluster);
+ doImport(cluster);
+
+ // All pending/ dirs should be empty, should have no SSTables left if all the transfers completed
+ assertPendingDirs(cluster, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty();
+ });
+
+ // Verify transfer IDs exist before compaction, then compact, then verify they're removed
+ assertCompaction(cluster, cluster, TRANSFERS_EXIST, TRANSFERS_EMPTY);
+
+ // Run after compaction, to enforce offset persistence + broadcast
+ assertSummary(cluster, summary -> {
+ Assertions.assertThat(summary).satisfies(s -> {
+ assert s.reconciledIds() == 1;
+ assert s.unreconciledIds() == 0;
+ });
+ });
+
+ assertLocalSelect(cluster, rows -> assertRows(rows, row(1, 1)));
+ }
+ }
+
+ @Test
+ public void importIndexAlreadyPresent() throws Throwable
+ {
+ try (Cluster cluster = cluster())
+ {
+ createSchema(cluster);
+
+ String indexName = "v_idx";
+ String indexCql = withKeyspace("CREATE INDEX " + indexName + " ON %s." + TABLE + " (v) USING 'sai'");
+ cluster.schemaChange(indexCql);
+
+ // This will add an SSTable that already has an SAI index, that needs to be distributed alongside the SSTable on transfer
+ IInvokableInstance importer = cluster.get(1);
+ long mark = importer.logs().mark();
+ doImport(cluster, importer, indexCql);
+ List<String> logs = importer.logs().grep(mark, "Submitting incremental index build of " + indexName).getResult();
+ Assertions.assertThat(logs).isEmpty();
+
+ // Index should exist and be queryable on all replicas after import
+ SAIUtil.assertIndexQueryable(cluster, KEYSPACE, indexName);
+
+ // Validate queries using the index
+ cluster.forEach(instance -> {
+ Object[][] rows = instance.executeInternal(withKeyspace("SELECT * FROM %s." + TABLE + " WHERE v = 1"));
+ assertRows(rows, row(1, 1));
+ });
+ }
+ }
+
+ @Test
+ public void importBuildsIndex() throws Throwable
+ {
+ try (Cluster cluster = cluster())
+ {
+ createSchema(cluster);
+
+ String indexName = "v_idx";
+ cluster.schemaChange(withKeyspace("CREATE INDEX " + indexName + " ON %s." + TABLE + " (v) USING 'sai'"));
+
+ // This will add an SSTable that's missing an SAI index, and the index will be built during the import on
+ // the coordinator
+ IInvokableInstance importer = cluster.get(1);
+ long mark = importer.logs().mark();
+ doImport(cluster, importer);
+ List<String> logs = importer.logs().watchFor(mark, Duration.ofMinutes(1), "Submitting incremental index build of " + indexName).getResult();
+ Assertions.assertThat(logs).isNotEmpty();
+
+ // Index should exist and be queryable on all replicas after import
+ SAIUtil.assertIndexQueryable(cluster, KEYSPACE, indexName);
+
+ // Validate queries using the index
+ cluster.forEach(instance -> {
+ Object[][] rows = instance.executeInternal(withKeyspace("SELECT * FROM %s." + TABLE + " WHERE v = 1"));
+ assertRows(rows, row(1, 1));
+ });
+ }
+ }
+
+ @Test
+ @Ignore
+ public void importReplicaDown() throws Throwable
+ {
+ try (Cluster cluster = cluster())
+ {
+ createSchema(cluster);
+
+ Iterable<IInvokableInstance> down = Collections.singleton(cluster.get(3));
+ Iterable<IInvokableInstance> up = cluster.stream().filter(instance -> instance != down).collect(Collectors.toList());
+ for (IInvokableInstance instance : down)
+ instance.shutdown().get();
+
+ doImport(cluster);
+
+ cluster.get(3).startup();
+
+ // Transfers did not complete, files should still exist on up replicas
+ assertPendingDirs(up, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isNotEmpty();
+ });
+ assertPendingDirs(down, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty();
+ });
+
+ // Transfers did not complete, transfer IDs should not be removed
+ assertCompaction(cluster, cluster, TRANSFERS_EXIST, TRANSFERS_EXIST);
+
+ assertLocalSelect(up, rows -> assertRows(rows, row(1, 1)));
+ }
+ }
+
+ @Test
+ public void importMissedActivationPrepare() throws Throwable
+ {
+ importMissedActivation(PREPARE);
+ }
+
+ @Test
+ public void importMissedActivationCommit() throws Throwable
+ {
+ importMissedActivation(COMMIT);
+ }
+
+ public void importMissedActivation(TransferActivation.Phase phase) throws Throwable
+ {
+ int MISSED_ACTIVATION = 2;
+ try (Cluster cluster = cluster(ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION)))
+ {
+ ByteBuddyInjections.SkipActivation.setup(cluster, phase);
+ createSchema(cluster);
+
+ Set<IInvokableInstance> missed = Collections.singleton(cluster.get(MISSED_ACTIVATION));
+ Iterable<IInvokableInstance> received = cluster.stream().filter(instance -> !missed.contains(instance)).collect(Collectors.toList());
+
+ Assertions.assertThatThrownBy(() -> doImport(cluster))
+ .hasMessageContaining("Failed adding SSTables")
+ .cause()
+ .hasMessageContaining("Failed streaming on 1 instance(s):")
+ .cause()
+ .hasMessageMatching("Tracked import failed during " + phase + " on " + cluster.get(MISSED_ACTIVATION).broadcastAddress() + " due to TIMEOUT")
+ .hasNoCause();
+
+ assertSummary(received, summary -> {
+ Assertions.assertThat(summary).satisfies(s -> {
+ assert s.reconciledIds() == 0;
+ assert s.unreconciledIds() == (phase == COMMIT ? 1 : 0);
+ });
+ });
+ assertSummary(missed, summary -> {
+ Assertions.assertThat(summary).satisfies(s -> {
+ assert s.reconciledIds() == 0;
+ assert s.unreconciledIds() == 0;
+ });
+ });
+
+ switch (phase)
+ {
+ case PREPARE:
+ // Activation did not start, files should be cleaned up
+ Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);
+ assertPendingDirs(cluster, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty();
+ });
+ break;
+ case COMMIT:
+ // Activation did not complete, files should still exist on all replicas
+ assertPendingDirs(cluster, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isNotEmpty();
+ });
+ break;
+ }
+
+ // If the activation is not everywhere, it shouldn't be purged on compaction
+ assertCompaction(cluster, received, TRANSFERS_EXIST, TRANSFERS_EXIST);
+
+ if (phase == PREPARE)
+ return;
+
+ // Permit activation of missed commits during read reconciliation
+ ByteBuddyInjections.SkipActivation.setup(cluster, null);
+
+ // Use coordinated query rather to confirm read reconciliation triggers activation
+ IInvokableInstance coordinator = cluster.get(3); // not initial transfer coordinator, but received activation
+ assertCoordinatedRead(coordinator, rows -> {
+ assertRows(rows, row(1, 1));
+ });
+
+ // Confirm others receive activation
+ assertLocalSelect(missed, rows -> {
+ assertRows(rows, row(1, 1));
+ });
+
+ assertCompaction(cluster, cluster, TRANSFERS_EXIST, TRANSFERS_EMPTY);
+
+ // Activation completed, files should be removed
+ assertPendingDirs(cluster, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty();
+ });
+ }
+ }
+
+ /**
+ * If a replica is missing an activation, executes a data read, then discovers the missing transfer during read
+ * reconciliation, it needs to augment the client response with the new transfer.
+ */
+ @Test
+ public void importMissingOnDataReplicaDuringAugment() throws Throwable
+ {
+ TransferActivation.Phase phase = COMMIT;
+ int MISSED_ACTIVATION = 2;
+ try (Cluster cluster = cluster(ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION)))
+ {
+ ByteBuddyInjections.SkipActivation.setup(cluster, phase);
+ createSchema(cluster);
+
+ IInvokableInstance missed = cluster.get(MISSED_ACTIVATION);
+
+ Assertions.assertThatThrownBy(() -> doImport(cluster))
+ .hasMessageContaining("Failed adding SSTables")
+ .cause()
+ .hasMessageContaining("Failed streaming on 1 instance(s):")
+ .cause()
+ .hasMessageMatching("Tracked import failed during " + phase + " on " + missed.broadcastAddress() + " due to TIMEOUT")
+ .hasNoCause();
+
+ assertSummary(Collections.singleton(missed), summary -> {
+ Assertions.assertThat(summary).satisfies(s -> {
+ assert s.reconciledIds() == 0;
+ assert s.unreconciledIds() == 0;
+ });
+ });
+
+ // Permit activation of missed commits during read reconciliation
+ ByteBuddyInjections.SkipActivation.setup(cluster, null);
+
+ // First read will fail due to failure to augment with transfer
+ long mark = missed.logs().mark();
+ Assertions.assertThatThrownBy(() -> {
+ assertCoordinatedRead(missed, rows -> {
+ assertRows(rows, row(1, 1));
+ });
+ }).isInstanceOf(missed.callOnInstance(() -> ReadTimeoutException.class)); // use instance classloader
+
+ List<String> logs = missed.logs().grep(mark, "Missing mutation ShortMutationId").getResult();
+ Assertions.assertThat(logs).isNotEmpty();
+
+ // Retry succeeds
+ assertCoordinatedRead(missed, rows -> {
+ assertRows(rows, row(1, 1));
+ });
+ }
+ }
+
+ /*
+ * When an import fails, bounce must not move the pending SSTables into the live set.
+ */
+ @Test
+ public void importBounceAfterPending() throws Throwable
+ {
+ IInstanceInitializer initializer = ByteBuddyInjections.SkipActivation.install(1, 2, 3);
+ try (Cluster cluster = cluster(initializer))
+ {
+ ByteBuddyInjections.SkipActivation.setup(cluster, COMMIT);
+ createSchema(cluster);
+
+ Assertions.assertThatThrownBy(() -> doImport(cluster))
+ .hasMessageContaining("Failed adding SSTables")
+ .cause()
+ .hasMessageContaining("Tracked import failed during COMMIT");
+
+ Runnable assertEmpty = () -> {
+ // Activation did not complete, files should still exist on all replicas
+ assertPendingDirs(cluster, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isNotEmpty();
+ });
+
+ // No one has activated, so should not be present in any summary
+ assertSummary(cluster, summary -> {
+ Assertions.assertThat(summary).satisfies(s -> {
+ assert s.reconciledIds() == 0;
+ assert s.unreconciledIds() == 0;
+ });
+ });
+
+ assertLocalSelect(cluster, rows -> assertRows(rows, EMPTY_ROWS));
+ };
+
+ assertEmpty.run();
+
+ bounce(cluster);
+
+ assertEmpty.run();
+ }
+ }
+
+ @Test
+ public void importOutOfRange() throws Throwable
+ {
+ try (Cluster cluster = cluster())
+ {
+ createSchema(cluster, 1);
+
+ Set<IInvokableInstance> inRange = new HashSet<>();
+ Set<IInvokableInstance> outOfRange = new HashSet<>();
+ cluster.forEach(instance -> {
+ boolean importReplica = instance.callOnInstance(() -> {
+ ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+ DataPlacement placement = ClusterMetadata.current().placements.get(cfs.keyspace.getMetadata().params.replication);
+ return placement.writes.forToken(IMPORT_TOKEN).get().containsSelf();
+ });
+ (importReplica ? inRange : outOfRange).add(instance);
+ });
+ logger.info("inRange: {}, outOfRange: {}", inRange, outOfRange);
+
+ Assertions.assertThat(inRange).hasSize(1);
+ IInvokableInstance onlyInRange = inRange.iterator().next();
+
+ // Reject import out of range
+ for (IInvokableInstance instance : outOfRange)
+ {
+ long mark = instance.logs().mark();
+ Consumer<List<String>> onResult = failedDirs -> Assertions.assertThat(failedDirs).hasSize(1);
+ doImport(cluster, instance, onResult, null);
+ instance.logs().grep(mark, "java.lang.RuntimeException: Key DecoratedKey(-4069959284402364209, 00000001) is not contained in the given ranges");
+ }
+
+ doImport(cluster, onlyInRange);
+
+ assertSummary(Collections.singleton(onlyInRange), summary -> {
+ Assertions.assertThat(summary).satisfies(s -> {
+ assert s.reconciledIds() == 1;
+ assert s.unreconciledIds() == 0;
+ });
+ });
+
+ for (IInvokableInstance instance : outOfRange)
+ {
+ // Out of range shouldn't have any transfers
+ assertCompaction(cluster, Collections.singleton(instance), TRANSFERS_EMPTY, TRANSFERS_EMPTY);
+
+ // Run after compaction, to enforce offset persistence + broadcast
+ assertSummary(Collections.singleton(instance), summary -> {
+ Assertions.assertThat(summary).isNull();
+ });
+ }
+ }
+ }
+
+ /*
+ * Ensure that activation IDs attached to SSTables aren't spread across Token boundaries by compaction.
+ *
+ * For example:
+ * IMPORT_TOKEN is owned by replicas (A, B)
+ * OUTSIDE_IMPORT_TOKEN is owned by replicas (B, C)
+ * Execute import so (A, B) have IMPORT_TOKEN
+ * Execute plain write so (B, C) have OUTSIDE_IMPORT_TOKEN
+ * Do a major compaction on B so IMPORT_TOKEN and OUTSIDE_IMPORT_TOKEN are compacted together into the same SSTable
+ * Execute a data read for OUTSIDE_IMPORT_TOKEN against B, ensure it doesn't contain any activation IDs
+ */
+ @Test
+ public void importActivationMergedByCompaction() throws Throwable
+ {
+ IInstanceInitializer initializer = (cl, tg, instance, gen) -> {
+ ByteBuddyInjections.SkipPurgeTransfers.install().initialise(cl, tg, instance, gen);
+ };
+ try (Cluster cluster = cluster(initializer))
+ {
+ createSchema(cluster, 2);
+
+ Set<IInvokableInstance> inImportRange = new HashSet<>();
+ cluster.forEach(instance -> {
+ logger.debug("Instance {} ring is {}", ClusterUtils.instanceId(instance), ClusterUtils.ring(instance));
+ boolean isInRange = instance.callOnInstance(() -> {
+ ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+ DataPlacement placement = ClusterMetadata.current().placements.get(cfs.keyspace.getMetadata().params.replication);
+ return placement.writes.forToken(IMPORT_TOKEN).get().containsSelf();
+ });
+ if (isInRange)
+ inImportRange.add(instance);
+ });
+ Assertions.assertThat(inImportRange).hasSize(2);
+
+ // Find a partition key that's not owned by the same replicas as the import
+ Murmur3Partitioner.LongToken NON_IMPORT_TOKEN = new Murmur3Partitioner.LongToken(IMPORT_TOKEN.getLongValue() * 3);
+ int NON_IMPORT_PK = Int32Type.instance.compose(Murmur3Partitioner.LongToken.keyForToken(NON_IMPORT_TOKEN));
+
+ Set<IInvokableInstance> inNonImportRange = new HashSet<>();
+ cluster.forEach(instance -> {
+ boolean isInRange = instance.callOnInstance(() -> {
+ ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+ DataPlacement placement = ClusterMetadata.current().placements.get(cfs.keyspace.getMetadata().params.replication);
+ return placement.writes.forToken(NON_IMPORT_TOKEN).get().containsSelf();
+ });
+ if (isInRange)
+ inNonImportRange.add(instance);
+ });
+ Assertions.assertThat(inNonImportRange).hasSize(2);
+ Assertions.assertThat(inNonImportRange).isNotEqualTo(inImportRange);
+
+ // Import: (A, B)
+ // Plain: (B, C)
+ IInvokableInstance A = null;
+ IInvokableInstance B = null;
+ IInvokableInstance C = null;
+ for (IInvokableInstance instance : cluster)
+ {
+ boolean isImport = inImportRange.contains(instance);
+ boolean isNonImport = inNonImportRange.contains(instance);
+ if (isImport && isNonImport)
+ B = instance;
+ else if (isImport)
+ A = instance;
+ else if (isNonImport)
+ C = instance;
+ }
+ Assertions.assertThat(A).isNotNull();
+ Assertions.assertThat(B).isNotNull();
+ Assertions.assertThat(C).isNotNull();
+
+ doImport(cluster, A);
+ assertLocalSelect(List.of(A, B), (IIsolatedExecutor.SerializableConsumer<Object[][]>) rows -> {
+ assertRows(rows, row(IMPORT_PK, IMPORT_PK));
+ });
+
+ ShortMutationId importTransferId = callSerialized(A, () -> ShortMutationId.serializer, () -> {
+ ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+ for (SSTableReader sstable : cfs.getLiveSSTables())
+ {
+ ActivatedTransfers transfers = sstable.getCoordinatorLogOffsets().transfers();
+ if (!transfers.isEmpty())
+ return transfers.iterator().next();
+ }
+ return null;
+ });
+ Assertions.assertThat(importTransferId).isNotNull();
+ C.coordinator().execute(withKeyspace("INSERT INTO %s." + TABLE + "(k, v) VALUES (?, ?)"), ConsistencyLevel.ALL, NON_IMPORT_PK, NON_IMPORT_PK);
+ assertCompaction(cluster, Collections.singleton(B), NOOP, NOOP);
+
+ // Reading from B for a range that doesn't include the import shouldn't include any transfer IDs, even though they've been compacted together
+ long mark = B.logs().mark();
+ Object[][] rows = B.coordinator().execute(withKeyspace("SELECT * FROM %s." + TABLE + " WHERE k = ?"), ConsistencyLevel.ALL, NON_IMPORT_PK);
+ assertRows(rows, row(NON_IMPORT_PK, NON_IMPORT_PK));
+ Assertions.assertThat(B.logs().grep(mark, "Found overlapping activation ID ").getResult()).isEmpty();
+
+ // But if the read range does include a transfer ID, it should have been added
+ mark = B.logs().mark();
+ rows = B.coordinator().execute(withKeyspace("SELECT * FROM %s." + TABLE + " WHERE k = ?"), ConsistencyLevel.ALL, IMPORT_PK);
+ assertRows(rows, row(IMPORT_PK, IMPORT_PK));
+ Assertions.assertThat(B.logs().grep(mark, "Found overlapping activation ID ").getResult()).isNotEmpty();
+ }
+ }
+
+ @Test
+ public void importFailedStreamCleanup() throws Throwable
+ {
+ int FAILED_STREAM = 3;
+ try (Cluster cluster = cluster(ByteBuddyInjections.FailIncomingStream.install(FAILED_STREAM)))
+ {
+ createSchema(cluster);
+
+ IInvokableInstance importer = cluster.get(1);
+ IInvokableInstance missed = cluster.get(FAILED_STREAM);
+
+ long mark = importer.logs().mark();
+ Assertions.assertThatThrownBy(() -> doImport(cluster, importer))
+ .isInstanceOf(RuntimeException.class)
+ .cause()
+ .isInstanceOf(RuntimeException.class)
+ .cause()
+ .hasMessageContaining("Remote peer " + missed.broadcastAddress() + " failed stream session");
+ List<String> logs = importer.logs().watchFor(mark, "Remote peer " + missed.broadcastAddress().toString() + " failed stream session").getResult();
+ Assertions.assertThat(logs).isNotEmpty();
+
+ // Await cleanup of failed stream
+ Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);
+
+ assertPendingDirs(cluster, (File pendingUuidDir) -> {
+ Assertions.assertThat(pendingUuidDir.listUnchecked(File::isFile)).isEmpty();
+ });
+
+ assertLocalSelect(cluster, rows -> {
+ assertRows(rows); // empty
+ });
+ }
+ }
+
+ public static class ByteBuddyInjections
+ {
+ // Only skips direct transfer activation, not activation as part of read reconciliation
+ public static class SkipActivation
+ {
+ // null to not skip
+ public static volatile TransferActivation.Phase phase;
+
+ public static IInstanceInitializer install(int...nodes)
+ {
+ return (ClassLoader cl, ThreadGroup tg, int num, int generation) -> {
+ for (int node : nodes)
+ if (node == num)
+ new ByteBuddy().rebase(TransferActivation.VerbHandler.class)
+ .method(named("doVerb"))
+ .intercept(MethodDelegation.to(ByteBuddyInjections.SkipActivation.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ };
+ }
+
+ // Need to set phase in each instance's classloader, otherwise assignment won't be visible to injected method body
+ public static void setup(Cluster cluster, TransferActivation.Phase phase)
+ {
+ logger.debug("Setting up phase {}", phase);
+ cluster.forEach(instance -> instance.runOnInstance(() -> ByteBuddyInjections.SkipActivation.phase = phase));
+ }
+
+ @SuppressWarnings("unused")
+ public static void doVerb(Message<TransferActivation> msg, @SuperCall Callable<?> zuper)
+ {
+ if (phase != null && msg.payload.phase == SkipActivation.phase)
+ {
+ logger.info("Skipping activation for test: {}", msg.payload);
+ return;
+ }
+
+ logger.info("Test running activation as usual: {}", msg.payload);
+
+ try
+ {
+ zuper.call();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ // ImmutableCoordinatorLogOffsets.Builder.purgeTransfers(Predicate)
+ public static class SkipPurgeTransfers
+ {
+ public static IInstanceInitializer install()
+ {
+ return (ClassLoader cl, ThreadGroup tg, int num, int generation) -> {
+ new ByteBuddy().rebase(ImmutableCoordinatorLogOffsets.Builder.class)
+ .method(named("purgeTransfers").and(takesArguments(Predicate.class)))
+ .intercept(MethodDelegation.to(SkipPurgeTransfers.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ };
+ }
+
+ @SuppressWarnings("unused")
+ public static void purgeTransfers()
+ {
+ logger.debug("Skipping purgeTransfers for test");
+ }
+ }
+
+ // CassandraStreamReceiver.finished
+ public static class FailIncomingStream
+ {
+ @SuppressWarnings("unused")
+ private static volatile boolean enabled = true;
+
+ public static IInstanceInitializer install(int... nodes)
+ {
+ return (ClassLoader cl, ThreadGroup tg, int num, int generation) -> {
+ for (int node : nodes)
+ if (node == num)
+ new ByteBuddy().rebase(CassandraStreamReceiver.class)
+ .method(named("finished").and(takesNoArguments()))
+ .intercept(MethodDelegation.to(FailIncomingStream.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ };
+ }
+
+ @SuppressWarnings("unused")
+ public static void finished(@This CassandraStreamReceiver self)
+ {
+ throw new RuntimeException("Failing incoming stream for test");
+ }
+
+ public static void toggle(Cluster cluster, boolean enable)
+ {
+ enabled = enable;
+ cluster.forEach(instance -> instance.runOnInstance(() -> FailIncomingStream.enabled = enable));
+ }
+ }
+ }
+
+ private static Cluster cluster() throws IOException
+ {
+ return cluster((cl, tg, instance, generation) -> {});
+ }
+
+ private static Cluster cluster(IInstanceInitializer initializer) throws IOException
+ {
+ return Cluster.build(NODES)
+ .withConfig(cfg -> cfg.with(Feature.NETWORK)
+ .with(Feature.GOSSIP)
+ .set("mutation_tracking_enabled", "true")
+ .set("write_request_timeout", "1000ms")
+ .set("autocompaction_on_startup_enabled", false)
+ .set("repair_request_timeout", "2s")
+ .set("stream_transfer_task_timeout", "10s"))
+ .withInstanceInitializer(initializer)
+ .start();
+ }
+
+ private static void createSchema(Cluster cluster)
+ {
+ createSchema(cluster, NODES);
+ }
+
+ private static void createSchema(Cluster cluster, int rf)
+ {
+ cluster.schemaChange(String.format(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
+ "{'class': 'SimpleStrategy', 'replication_factor': " + rf + "} " +
+ "AND replication_type='tracked';")));
+ cluster.schemaChange(TABLE_SCHEMA_CQL);
+ }
+
+ private static void doImport(Cluster cluster) throws IOException
+ {
+ doImport(cluster, cluster.get(1));
+ }
+
+ private static void doImport(Cluster cluster, IInvokableInstance target) throws IOException
+ {
+ doImport(cluster, target, failedDirs -> Assertions.assertThat(failedDirs).isEmpty(), null);
+ }
+
+ private static void doImport(Cluster cluster, IInvokableInstance target, @Nullable String createIndexCql) throws IOException
+ {
+ doImport(cluster, target, failedDirs -> Assertions.assertThat(failedDirs).isEmpty(), createIndexCql);
+ }
+
+ private static void doImport(Cluster cluster, IInvokableInstance target, Consumer<List<String>> onFailedDirs, @Nullable String createIndexCql) throws IOException
+ {
+ String file = Files.createTempDirectory(MutationTrackingTest.class.getSimpleName()).toString();
+
+ // Needs to run outside of instance executor because creates schema
+ CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder()
+ .forTable(TABLE_SCHEMA_CQL)
+ .inDirectory(file)
+ .using("INSERT INTO " + KEYSPACE_TABLE + " (k, v) " + "VALUES (?, ?)");
+
+ if (createIndexCql != null)
+ {
+ builder.withIndexes(createIndexCql).withBuildIndexes(true);
+ }
+
+ try (CQLSSTableWriter writer = builder.build())
+ {
+ writer.addRow(IMPORT_PK, 1);
+ }
+
+ assertLocalSelect(cluster, rows -> {
+ assertRows(rows); // empty
+ });
+
+ List<String> failed = target.callOnInstance(() -> {
+ ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+ Set<String> paths = Set.of(file);
+ logger.info("Importing SSTables {}", paths);
+ return cfs.importNewSSTables(paths, true, true, true, true, true, true, true);
+ });
+
+ // Sleep for a while to make sure import completes
+ Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
+ onFailedDirs.accept(failed);
+ }
+
+ private static void assertCoordinatedRead(IInvokableInstance instance, IIsolatedExecutor.SerializableConsumer<Object[][]> onRows)
+ {
+ ICoordinator coordinator = instance.coordinator();
+ String cql = "SELECT * FROM %s." + TABLE + " WHERE k = 1";
+ Object[][] rows = coordinator.execute(withKeyspace(cql), ConsistencyLevel.ALL);
+ onRows.accept(rows);
+ }
+
+ private static void assertPendingDirs(Iterable<IInvokableInstance> validate, IIsolatedExecutor.SerializableConsumer<File> forPendingUuidDir)
+ {
+ for (IInvokableInstance instance : validate)
+ {
+ instance.runOnInstance(() -> {
+ Set<File> allPendingDirs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE).getDirectories().getPendingLocations();
+ for (File pendingDir : allPendingDirs)
+ {
+ File[] pendingUuidDirs = pendingDir.listUnchecked(File::isDirectory);
+ for (File pendingUuidDir : pendingUuidDirs)
+ {
+ forPendingUuidDir.accept(pendingUuidDir);
+ }
+ }
+ });
+ }
+ }
+
+ private static void assertSummary(Iterable<IInvokableInstance> validate, IIsolatedExecutor.SerializableConsumer<MutationSummary> onSummary)
+ {
+ for (IInvokableInstance instance : validate)
+ {
+ instance.runOnInstance(() -> {
+ DecoratedKey key = DatabaseDescriptor.getPartitioner().decorateKey(ByteBufferUtil.bytes(1));
+ TableId tableId = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE).metadata().id;
+ MutationSummary summary;
+ try
+ {
+ summary = MutationTrackingService.instance.createSummaryForKey(key, tableId, false);
+ }
+ catch (UnknownShardException e)
+ {
+ summary = null;
+ }
+ logger.debug("Validating summary {}", summary);
+ onSummary.accept(summary);
+ });
+ }
+ }
+
+ private static void assertCompaction(Cluster cluster, Iterable<IInvokableInstance> validate,
+ IIsolatedExecutor.SerializableConsumer<SSTableReader> before,
+ IIsolatedExecutor.SerializableConsumer<SSTableReader> after)
+ {
+ for (IInvokableInstance instance : validate)
+ {
+ instance.runOnInstance(() -> {
+ ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+ for (SSTableReader sstable : cfs.getLiveSSTables())
+ {
+ logger.info("SSTable {} before compaction: {}", sstable.getFilename(), sstable.getCoordinatorLogOffsets());
+ before.accept(sstable);
+ }
+ });
+ }
+
+ // Activation ID must be persisted and broadcast across all peers in the cluster for any to mark as persisted + reconciled
+ cluster.forEach(i -> {
+ i.runOnInstance(() -> {
+ MutationTrackingService.instance.persistLogStateForTesting();
+ MutationTrackingService.instance.broadcastOffsetsForTesting();
+ });
+ });
+
+ // Broadcast is async, wait until completion
+ Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
+
+ for (IInvokableInstance instance : validate)
+ {
+ instance.runOnInstance(() -> {
+ ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(KEYSPACE, TABLE);
+ logger.info("Triggering compaction on instance {}", cfs.metadata.keyspace);
+ CompactionManager.instance.performMaximal(cfs);
+
+ for (SSTableReader sstable : cfs.getLiveSSTables())
+ {
+ logger.info("SSTable {} after compaction: {}", sstable.getFilename(), sstable.getCoordinatorLogOffsets());
+ after.accept(sstable);
+ }
+ });
+ }
+ }
+
+ private static void assertLocalSelect(Iterable<IInvokableInstance> validate, IIsolatedExecutor.SerializableConsumer<Object[][]> onRows)
+ {
+ for (IInvokableInstance instance : validate)
+ {
+ {
+ Object[][] rows = instance.executeInternal(withKeyspace("SELECT * FROM %s." + TABLE + " WHERE k = 1"));
+ onRows.accept(rows);
+ }
+ {
+ Object[][] rows = instance.executeInternal(withKeyspace("SELECT * FROM %s." + TABLE));
+ onRows.accept(rows);
+ }
+ }
+ }
+
+ private static void bounce(Cluster cluster)
+ {
+ cluster.forEach(instance -> {
+ try
+ {
+ instance.shutdown().get();
+ }
+ catch (InterruptedException | ExecutionException e)
+ {
+ throw new RuntimeException(e);
+ }
+ instance.startup();
+ });
+ }
+
+ private static <T> T callSerialized(IInvokableInstance instance, IIsolatedExecutor.SerializableSupplier<IVersionedSerializer<T>> serializer, IIsolatedExecutor.SerializableCallable<T> callable)
+ {
+ ByteBuffer serialized = instance.callOnInstance(() -> {
+ T deserialized = callable.call();
+ IVersionedSerializer<T> serialize = serializer.get();
+ try
+ {
+ return serialize.serialize(deserialized, MessagingService.current_version);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ });
+ try
+ {
+ return serializer.get().deserialize(serialized, MessagingService.current_version);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java
index ebc5bf1df5..84e42694aa 100644
--- a/test/unit/org/apache/cassandra/ServerTestUtils.java
+++ b/test/unit/org/apache/cassandra/ServerTestUtils.java
@@ -50,6 +50,7 @@ import org.apache.cassandra.locator.BaseProximity;
import org.apache.cassandra.locator.Endpoint;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.service.DiskErrorsHandlerService;
import org.apache.cassandra.service.EmbeddedCassandraService;
@@ -154,6 +155,9 @@ public final class ServerTestUtils
{
daemonInitialization();
+ // Need to happen after daemonInitialization for config to be set, but before CFS initialization
+ MutationJournal.instance.start();
+
if (isServerPrepared)
return;
diff --git a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
index 20d098fcb1..b19cc1190e 100644
--- a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
+++ b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
@@ -30,7 +30,6 @@ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.replication.MutationId;
-import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.MemtableParams;
@@ -60,8 +59,8 @@ public class CoordinatorLogOffsetsLifecycleTest
static
{
+ // Needs to happen before memtableConfigs is called, since that's populated from loaded Config
DatabaseDescriptor.daemonInitialization();
- MutationJournal.instance.start();
}
@BeforeClass
@@ -180,8 +179,8 @@ public class CoordinatorLogOffsetsLifecycleTest
Memtable memtable = view.getCurrentMemtable();
ImmutableCoordinatorLogOffsets logOffsets = memtable.getFlushSet(null, null).coordinatorLogOffsets();
- Assertions.assertThat(logOffsets.size()).isEqualTo(1);
- Assertions.assertThat(logOffsets.offsets(id2.logId()).contains(id2.offset())).isTrue();
+ Assertions.assertThat(logOffsets.mutations().size()).isEqualTo(1);
+ Assertions.assertThat(logOffsets.mutations().offsets(id2.logId()).contains(id2.offset())).isTrue();
}
// flush 1
@@ -195,8 +194,8 @@ public class CoordinatorLogOffsetsLifecycleTest
SSTableReader sstable = Iterables.getOnlyElement(view.liveSSTables());
ImmutableCoordinatorLogOffsets logOffsets = sstable.getCoordinatorLogOffsets();
- Assertions.assertThat(logOffsets.size()).isEqualTo(1);
- Assertions.assertThat(logOffsets.offsets(id2.logId()).contains(id2.offset())).isTrue();
+ Assertions.assertThat(logOffsets.mutations().size()).isEqualTo(1);
+ Assertions.assertThat(logOffsets.mutations().offsets(id2.logId()).contains(id2.offset())).isTrue();
// Single-participant, so mutations are immediately reconciled once applied
Assertions.assertThat(sstable.isRepaired()).isTrue();
}
@@ -214,8 +213,8 @@ public class CoordinatorLogOffsetsLifecycleTest
Memtable memtable = view.getCurrentMemtable();
ImmutableCoordinatorLogOffsets logOffsets = memtable.getFlushSet(null, null).coordinatorLogOffsets();
- Assertions.assertThat(logOffsets.size()).isEqualTo(1);
- Assertions.assertThat(logOffsets.offsets(id4.logId()).contains(id4.offset())).isTrue();
+ Assertions.assertThat(logOffsets.mutations().size()).isEqualTo(1);
+ Assertions.assertThat(logOffsets.mutations().offsets(id4.logId()).contains(id4.offset())).isTrue();
}
// flush 2
@@ -231,14 +230,14 @@ public class CoordinatorLogOffsetsLifecycleTest
sstables.sort(Comparator.comparing(sst -> sst.descriptor.id.asBytes()));
{
ImmutableCoordinatorLogOffsets logOffsets = sstables.get(0).getCoordinatorLogOffsets();
- Assertions.assertThat(logOffsets.size()).isEqualTo(1);
- Assertions.assertThat(logOffsets.offsets(id2.logId()).contains(id2.offset())).isTrue();
+ Assertions.assertThat(logOffsets.mutations().size()).isEqualTo(1);
+ Assertions.assertThat(logOffsets.mutations().offsets(id2.logId()).contains(id2.offset())).isTrue();
}
{
ImmutableCoordinatorLogOffsets logOffsets = sstables.get(1).getCoordinatorLogOffsets();
- Assertions.assertThat(logOffsets.size()).isEqualTo(1);
- Assertions.assertThat(logOffsets.offsets(id4.logId()).contains(id4.offset())).isTrue();
+ Assertions.assertThat(logOffsets.mutations().size()).isEqualTo(1);
+ Assertions.assertThat(logOffsets.mutations().offsets(id4.logId()).contains(id4.offset())).isTrue();
}
for (SSTableReader sstable : sstables)
Assertions.assertThat(sstable.isRepaired()).isTrue();
@@ -254,8 +253,8 @@ public class CoordinatorLogOffsetsLifecycleTest
SSTableReader sstable = Iterables.getOnlyElement(view.liveSSTables());
ImmutableCoordinatorLogOffsets logOffsets = sstable.getCoordinatorLogOffsets();
- Assertions.assertThat(logOffsets.size()).isEqualTo(1);
- Assertions.assertThat(logOffsets.offsets(id4.logId()).contains(id4.offset())).isTrue();
+ Assertions.assertThat(logOffsets.mutations().size()).isEqualTo(1);
+ Assertions.assertThat(logOffsets.mutations().offsets(id4.logId()).contains(id4.offset())).isTrue();
Assertions.assertThat(sstable.isRepaired()).isTrue();
}
}
diff --git a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java
index 1404953743..98f297663b 100644
--- a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java
@@ -133,6 +133,6 @@ public class AntiCompactionBytemanTest extends CQLTester
t.join();
assertFalse(failed.get());
assertFalse(getCurrentColumnFamilyStore().getLiveSSTables().contains(sstableBefore));
- Util.assertOnDiskState(getCurrentColumnFamilyStore(), 3);
+ Util.assertOnDiskState(getCurrentColumnFamilyStore(), 2);
}
}
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
index e4d99f295a..57e76c9a46 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java
@@ -63,7 +63,7 @@ import static java.util.Collections.singleton;
public class TrackerTest
{
- private static final int COMPONENT_STATS_SIZE_BYTES = 4805;
+ private static final int COMPONENT_STATS_SIZE_BYTES = 4806;
private static final class MockListener implements INotificationConsumer
{
diff --git a/test/unit/org/apache/cassandra/replication/ActivatedTransfersTest.java b/test/unit/org/apache/cassandra/replication/ActivatedTransfersTest.java
new file mode 100644
index 0000000000..b309e92618
--- /dev/null
+++ b/test/unit/org/apache/cassandra/replication/ActivatedTransfersTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import accord.utils.Gen;
+import accord.utils.Gens;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.BufferDecoratedKey;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Bounds;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.AccordGenerators;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.CassandraGenerators;
+import org.assertj.core.api.Assertions;
+
+import static accord.utils.Property.qt;
+
+public class ActivatedTransfersTest
+{
+ @BeforeClass
+ public static void setUp()
+ {
+ DatabaseDescriptor.daemonInitialization();
+ DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+ }
+
+ private static Token tk(long token)
+ {
+ return new Murmur3Partitioner.LongToken(token);
+ }
+
+ private static PartitionPosition pos(long token)
+ {
+ Token t = tk(token);
+ return new BufferDecoratedKey(t, ByteBufferUtil.bytes(token));
+ }
+
+ private static Bounds<PartitionPosition> bounds(long left, long right)
+ {
+ return (Bounds) bounds(left, true, right, true);
+ }
+
+ private static Range<PartitionPosition> range(long left, long right)
+ {
+ return (Range) bounds(left, false, right, true);
+ }
+
+ private static AbstractBounds<PartitionPosition> bounds(long left, boolean leftInclusive, long right, boolean rightInclusive)
+ {
+ return AbstractBounds.bounds(pos(left), leftInclusive, pos(right), rightInclusive);
+ }
+
+ private static ShortMutationId id(int logId, int offset)
+ {
+ return new ShortMutationId(logId, offset);
+ }
+
+ private static Gen<Token> tokenGen()
+ {
+ return AccordGenerators.fromQT(CassandraGenerators.murmurToken());
+ }
+
+ private static Gen<CoordinatorLogId> logIdGen()
+ {
+ return rs -> new CoordinatorLogId(rs.nextInt(), rs.nextInt());
+ }
+
+ private static Gen<ShortMutationId> idGen()
+ {
+ return rs -> {
+ int offset = (short) rs.nextInt(Short.MIN_VALUE, Short.MAX_VALUE);
+ return new ShortMutationId(logIdGen().next(rs).asLong(), offset);
+ };
+ }
+
+ private static Gen<ActivatedTransfers> activatedTransfersGen()
+ {
+ return rs -> {
+ List<ActivatedTransfers.ActivatedTransfer> entries = Gens.lists(activatedTransferGen()).ofSizeBetween(0, 10).next(rs);
+ ActivatedTransfers transfers = new ActivatedTransfers();
+ entries.forEach(entry -> transfers.add(entry.id, entry.bounds));
+ return transfers;
+ };
+ }
+
+ private static Gen<ActivatedTransfers.ActivatedTransfer> activatedTransferGen()
+ {
+ return rs -> {
+ ShortMutationId id = idGen().next(rs);
+ while (true)
+ {
+ Token left = tokenGen().next(rs);
+ Token right = tokenGen().next(rs);
+
+ if (!AbstractBounds.strictlyWrapsAround(left, right))
+ return new ActivatedTransfers.ActivatedTransfer(id, new Bounds<>(left, right));
+ }
+ };
+ }
+
+ @Test
+ public void testSerdeRoundtrip()
+ {
+ qt()
+ .forAll(activatedTransfersGen())
+ .check(transfers -> {
+ int version = MessagingService.current_version;
+ ActivatedTransfers deserialized;
+ try (DataOutputBuffer out = new DataOutputBuffer())
+ {
+ ActivatedTransfers.serializer.serialize(transfers, out, version);
+
+ try (DataInputBuffer in = new DataInputBuffer(out.buffer(), true))
+ {
+ deserialized = ActivatedTransfers.serializer.deserialize(in, version);
+ }
+ }
+
+ Assertions.assertThat(deserialized).isEqualTo(transfers);
+ });
+ }
+
+ @Test
+ public void testIntersectsSingle()
+ {
+ ActivatedTransfers transfers = new ActivatedTransfers();
+ ShortMutationId id1 = id(1, 0);
+ transfers.add(id1, new Bounds<>(tk(400), tk(500))); // [400, 500]
+
+ final long min = Murmur3Partitioner.instance.getMinimumToken().token;
+
+ // Token
+ assertNoIntersection(transfers, tk(0));
+ assertIntersects(transfers, id1, tk(400));
+ assertIntersects(transfers, id1, tk(450));
+ assertIntersects(transfers, id1, tk(500));
+ assertNoIntersection(transfers, tk(550));
+
+ // Bounds []
+ assertNoIntersection(transfers, bounds(100, 300)); // [100, 300]
+ assertIntersects(transfers, id1, bounds(100, 400L)); // [100, 400] - overlap at boundary
+ assertIntersects(transfers, id1, bounds(100, 450L)); // [100, 450]
+ assertIntersects(transfers, id1, bounds(400, 500L)); // [400, 500]
+ assertIntersects(transfers, id1, bounds(500, 600)); // [500, 600] - overlap at boundary
+ assertNoIntersection(transfers, bounds(600, 700)); // [600, 700]
+ assertIntersects(transfers, id1, bounds(0, 1000)); // [0, 1000]
+ assertIntersects(transfers, id1, bounds(400, 400)); // [400, 400]
+ assertIntersects(transfers, id1, new Bounds<>(tk(500).minKeyBound(), tk(min).maxKeyBound())); // [400, 400]
+
+ // Range (]
+ assertNoIntersection(transfers, range(100, 300)); // (100, 300]
+ assertIntersects(transfers, id1, range(100, 400L)); // (100, 400] - overlap at boundary
+ assertIntersects(transfers, id1, range(100, 450L)); // (100, 450]
+ assertIntersects(transfers, id1, range(400, 500L)); // (400, 500]
+ assertNoIntersection(transfers, range(500, 600)); // (500, 600]
+ assertNoIntersection(transfers, range(600, 700)); // (600, 700]
+ assertIntersects(transfers, id1, range(0, 1000)); // (0, 1000]
+ assertIntersects(transfers, id1, range(0, min)); // (0, MIN]
+ assertIntersects(transfers, id1, range(450, min)); // (450, MIN]
+ assertIntersects(transfers, id1, range(0, 0)); // (0, 0]
+ assertNoIntersection(transfers, range(600, 300)); // (600, 300]
+
+ // [)
+ assertNoIntersection(transfers, bounds(100, true, 300, false)); // [100, 300)
+ assertNoIntersection(transfers, bounds(100, true, 400, false)); // [100, 400)
+ assertIntersects(transfers, id1, bounds(100, true, 450, false)); // [100, 450)
+ assertIntersects(transfers, id1, bounds(400, true, 500, false)); // [400, 500)
+ assertIntersects(transfers, id1, bounds(500, true, 600, false)); // [500, 600)
+ assertNoIntersection(transfers, bounds(600, true, 700, false)); // [600, 700)
+ assertIntersects(transfers, id1, bounds(0, true, 1000, false)); // [0, 1000)
+
+ // ()
+ assertNoIntersection(transfers, bounds(100, false, 300, false)); // (100, 300)
+ assertNoIntersection(transfers, bounds(100, false, 400, false)); // (100, 400)
+ assertIntersects(transfers, id1, bounds(100, false, 450, false)); // (100, 450)
+ assertIntersects(transfers, id1, bounds(400, false, 500, false)); // (400, 500)
+ assertNoIntersection(transfers, bounds(500, false, 600, false)); // (500, 600)
+ assertNoIntersection(transfers, bounds(600, false, 700, false)); // (600, 700)
+ assertIntersects(transfers, id1, bounds(0, false, 1000, false)); // (0, 1000)
+ }
+
+ @Test
+ public void testIntersectsMultiple()
+ {
+ ActivatedTransfers transfers = new ActivatedTransfers();
+ ShortMutationId id1 = id(100, 1);
+ ShortMutationId id2 = id(100, 2);
+ ShortMutationId id3 = id(100, 3);
+
+ transfers.add(id1, new Bounds<>(tk(100), tk(200)));
+ transfers.add(id2, new Bounds<>(tk(300), tk(400)));
+ transfers.add(id3, new Bounds<>(tk(500), tk(600)));
+
+ Set<ShortMutationId> ids1 = new HashSet<>();
+ transfers.forEachIntersecting(new Bounds<>(pos(50), pos(150)), ids1::add);
+ Assertions.assertThat(ids1).containsExactly(id1);
+
+ Set<ShortMutationId> ids2 = new HashSet<>();
+ transfers.forEachIntersecting(new Bounds<>(pos(50), pos(350)), ids2::add);
+ Assertions.assertThat(ids2).containsExactly(id1, id2);
+
+ Set<ShortMutationId> ids3 = new HashSet<>();
+ transfers.forEachIntersecting(new Bounds<>(pos(0), pos(700)), ids3::add);
+ Assertions.assertThat(ids3).containsExactly(id1, id2, id3);
+ }
+
+ @Test
+ public void testRemove()
+ {
+ ActivatedTransfers transfers = new ActivatedTransfers();
+ ShortMutationId id1 = id(100, 1);
+ ShortMutationId id2 = id(100, 2);
+ ShortMutationId id3 = id(100, 3);
+
+ transfers.add(id1, new Bounds<>(tk(100), tk(200)));
+ transfers.add(id2, new Bounds<>(tk(300), tk(400)));
+ transfers.add(id3, new Bounds<>(tk(500), tk(600)));
+
+ transfers.removeOffset(1);
+
+ assertNoIntersection(transfers, bounds(50, 100));
+ }
+
+ private void assertIntersects(ActivatedTransfers transfers, ShortMutationId expectedId, Token token)
+ {
+ Set<ShortMutationId> ids = new HashSet<>();
+ transfers.forEachIntersecting(token, ids::add);
+ Assertions.assertThat(ids).contains(expectedId);
+ }
+
+ private void assertIntersects(ActivatedTransfers transfers, ShortMutationId expectedId, AbstractBounds<PartitionPosition> range)
+ {
+ Set<ShortMutationId> ids = new HashSet<>();
+ transfers.forEachIntersecting(range, ids::add);
+ Assertions.assertThat(ids).contains(expectedId);
+ }
+
+ private void assertNoIntersection(ActivatedTransfers transfers, Token token)
+ {
+ Set<ShortMutationId> ids = new HashSet<>();
+ transfers.forEachIntersecting(token, ids::add);
+ Assertions.assertThat(ids).isEmpty();
+ }
+
+ private void assertNoIntersection(ActivatedTransfers transfers, AbstractBounds<PartitionPosition> range)
+ {
+ Set<ShortMutationId> ids = new HashSet<>();
+ transfers.forEachIntersecting(range, ids::add);
+ Assertions.assertThat(ids).isEmpty();
+ }
+}
diff --git a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
index 253d81526b..f2d6387bfb 100644
--- a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
+++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
@@ -22,6 +22,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@@ -82,10 +83,13 @@ public class CoordinatorLogOffsetsTest
private static final Gen<MutationId> MUTATION_ID_GEN = rs -> new MutationId(LOG_ID_GEN.next(rs), SEQUENCE_ID_GEN.next(rs));
private static final Gen<ImmutableCoordinatorLogOffsets> COORDINATOR_LOG_OFFSETS_GEN = rs -> {
- int numIds = rs.nextBiasedInt(0, 10, 1000);
- ImmutableCoordinatorLogOffsets.Builder builder = new ImmutableCoordinatorLogOffsets.Builder(numIds);
- for (int i = 0; i < numIds; i++)
+ int numMutations = rs.nextBiasedInt(0, 10, 1000);
+ ImmutableCoordinatorLogOffsets.Builder builder = new ImmutableCoordinatorLogOffsets.Builder(numMutations);
+ for (int i = 0; i < numMutations; i++)
builder.add(MUTATION_ID_GEN.next(rs));
+ int numTransfers = rs.nextBiasedInt(0, 1, 10);
+ for (int i = 0; i < numTransfers; i++)
+ builder.addTransfer(MUTATION_ID_GEN.next(rs), new Bounds<>(new Murmur3Partitioner.LongToken(1), new Murmur3Partitioner.LongToken(2)));
return builder.build();
};
@@ -100,18 +104,18 @@ public class CoordinatorLogOffsetsTest
{
qt()
.forAll(COORDINATOR_LOG_OFFSETS_GEN)
- .check(offsets -> {
+ .check(originals -> {
try (DataOutputBuffer outputBuffer = DataOutputBuffer.scratchBuffer.get())
{
- ImmutableCoordinatorLogOffsets.serializer.serialize(offsets, outputBuffer, MessagingService.current_version);
- byte[] bytes = outputBuffer.toByteArray();
- try (DataInputBuffer inputBuffer = new DataInputBuffer(bytes))
+ ImmutableCoordinatorLogOffsets.serializer.serialize(originals, outputBuffer, MessagingService.current_version);
+ try (DataInputBuffer inputBuffer = new DataInputBuffer(outputBuffer.buffer(), true))
{
ImmutableCoordinatorLogOffsets deserialized = ImmutableCoordinatorLogOffsets.serializer.deserialize(inputBuffer, MessagingService.current_version);
- Assertions.assertThat(Sets.newHashSet(deserialized.iterator())).isEqualTo(Sets.newHashSet(offsets.iterator()));
- for (long logId : offsets)
- Assertions.assertThat(deserialized.offsets(logId)).isEqualTo(offsets.offsets(logId));
- Assertions.assertThat(bytes.length).isEqualTo(ImmutableCoordinatorLogOffsets.serializer.serializedSize(offsets, MessagingService.current_version));
+ CoordinatorLogOffsets.Mutations<Offsets.Immutable> mutations = deserialized.mutations();
+ Assertions.assertThat(Sets.newHashSet(mutations)).isEqualTo(Sets.newHashSet(originals.mutations().iterator()));
+ for (long logId : originals.mutations())
+ Assertions.assertThat(mutations.offsets(logId)).isEqualTo(originals.mutations().offsets(logId));
+ Assertions.assertThat(outputBuffer.getLength()).isEqualTo(ImmutableCoordinatorLogOffsets.serializer.serializedSize(originals, MessagingService.current_version));
}
}
});
@@ -132,10 +136,10 @@ public class CoordinatorLogOffsetsTest
MutableCoordinatorLogOffsets logOffsets = ctor.get();
for (MutationId id : ids)
{
- Offsets originalOffsets = logOffsets.offsets(id.logId());
+ Offsets originalOffsets = logOffsets.mutations().offsets(id.logId());
boolean existed = originalOffsets.contains(id.offset());
logOffsets.add(id);
- Offsets updatedOffsets = logOffsets.offsets(id.logId());
+ Offsets updatedOffsets = logOffsets.mutations().offsets(id.logId());
if (existed)
Assertions.assertThat(updatedOffsets).hasSameSizeAs(originalOffsets);
Assertions.assertThat(updatedOffsets.contains(id.offset())).isTrue();
@@ -156,13 +160,13 @@ public class CoordinatorLogOffsetsTest
.forAll(COORDINATOR_LOG_OFFSETS_GEN, COORDINATOR_LOG_OFFSETS_GEN)
.check((left, right) -> {
MutableCoordinatorLogOffsets merged = ctor.get();
- merged.addAll(left);
- merged.addAll(right);
- for (Long logId : merged)
+ merged.addAll(left.mutations());
+ merged.addAll(right.mutations());
+ for (Long logId : merged.mutations())
{
- Offsets leftOffsets = left.offsets(logId);
- Offsets rightOffsets = right.offsets(logId);
- Offsets mergedOffsets = Offsets.Immutable.copy(merged.offsets(logId));
+ Offsets leftOffsets = left.mutations().offsets(logId);
+ Offsets rightOffsets = right.mutations().offsets(logId);
+ Offsets mergedOffsets = Offsets.Immutable.copy(merged.mutations().offsets(logId));
Assertions.assertThat(mergedOffsets).isEqualTo(Offsets.Immutable.union(leftOffsets, rightOffsets));
}
});
@@ -189,9 +193,9 @@ public class CoordinatorLogOffsetsTest
}
ImmutableCoordinatorLogOffsets fromBuilder = builder.build();
- Assertions.assertThat(fromBuilder).hasSize(logOffsets.size());
- for (Long logId : logOffsets)
- Assertions.assertThat(fromBuilder.offsets(logId)).isEqualTo(Offsets.Immutable.copy(logOffsets.offsets(logId)));
+ Assertions.assertThat(fromBuilder.mutations()).hasSize(logOffsets.mutations().size());
+ for (Long logId : logOffsets.mutations())
+ Assertions.assertThat(fromBuilder.mutations().offsets(logId)).isEqualTo(Offsets.Immutable.copy(logOffsets.mutations().offsets(logId)));
});
}
@@ -232,9 +236,9 @@ public class CoordinatorLogOffsetsTest
for (Future<?> task : concurrentUpdates)
task.get();
- Assertions.assertThatIterable(exclusive).hasSameSizeAs(concurrent);
- for (Long logId : exclusive)
- Assertions.assertThat(exclusive.offsets(logId)).isEqualTo(concurrent.offsets(logId));
+ Assertions.assertThatIterable(exclusive.mutations()).hasSameSizeAs(concurrent.mutations());
+ for (Long logId : exclusive.mutations())
+ Assertions.assertThat(exclusive.mutations().offsets(logId)).isEqualTo(concurrent.mutations().offsets(logId));
});
}
@@ -289,7 +293,7 @@ public class CoordinatorLogOffsetsTest
.add(mutation.id())
.build();
Range<Token> range = getShardRange(mutation);
- List<? extends Offsets> offsets = Collections.singletonList(logOffsets.offsets(mutation.id().logId()));
+ List<? extends Offsets> offsets = Collections.singletonList(logOffsets.mutations().offsets(mutation.id().logId()));
MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr2);
MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr3);
@@ -323,7 +327,7 @@ public class CoordinatorLogOffsetsTest
.build();
Range<Token> range = getShardRange(mutation);
- List<? extends Offsets> offsets = Collections.singletonList(logOffsets.offsets(mutation.id().logId()));
+ List<? extends Offsets> offsets = Collections.singletonList(logOffsets.mutations().offsets(mutation.id().logId()));
MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr2);
MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr3);
diff --git a/test/unit/org/apache/cassandra/replication/LocalTransfersTest.java b/test/unit/org/apache/cassandra/replication/LocalTransfersTest.java
new file mode 100644
index 0000000000..465f280c49
--- /dev/null
+++ b/test/unit/org/apache/cassandra/replication/LocalTransfersTest.java
@@ -0,0 +1,344 @@
+/*
+ * 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.Collection;
+import java.util.Collections;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.TimeUUID;
+import org.assertj.core.api.Assertions;
+
+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.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.utils.TimeUUID.Generator.nextTimeUUID;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+public class LocalTransfersTest
+{
+ private LocalTransfers localTransfers;
+ private ShortMutationId transferId;
+ private TimeUUID planId;
+
+ @BeforeClass
+ public static void setUpClass()
+ {
+ DatabaseDescriptor.daemonInitialization();
+ DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+ }
+
+ @Before
+ public void setUp() throws Exception
+ {
+ localTransfers = new LocalTransfers();
+ transferId = new ShortMutationId(1, 100);
+ planId = nextTimeUUID();
+ }
+
+ private CoordinatedTransfer coordinatedTransfer(ShortMutationId transferId)
+ {
+ return coordinatedTransfer(transferId, new Range<>(tk(0), tk(1000)));
+ }
+
+ private CoordinatedTransfer coordinatedTransfer(ShortMutationId transferId, Range<Token> range)
+ {
+ MutationId mutationId = transferId != null
+ ? new MutationId(transferId.logId(), transferId.offset(), (int) System.currentTimeMillis())
+ : null;
+ return new CoordinatedTransfer(range, mutationId);
+ }
+
+ private PendingLocalTransfer pendingTransfer(TimeUUID planId)
+ {
+ SSTableReader mockSSTable = mock(SSTableReader.class);
+ Collection<SSTableReader> sstables = Collections.singletonList(mockSSTable);
+
+ return new PendingLocalTransfer(planId, sstables);
+ }
+
+ private static Token tk(long token)
+ {
+ return new Murmur3Partitioner.LongToken(token);
+ }
+
+ @Test
+ public void testSaveCoordinatedTransfer()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ localTransfers.save(transfer);
+ localTransfers.activating(transfer);
+
+ CoordinatedTransfer loaded = localTransfers.getActivatedTransfer(transferId);
+ Assertions.assertThat(loaded).isEqualTo(transfer);
+
+ assertThatThrownBy(() -> localTransfers.save(transfer))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ public void testActivatingTransfer()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ localTransfers.save(transfer);
+ localTransfers.activating(transfer);
+
+ CoordinatedTransfer retrieved = localTransfers.getActivatedTransfer(transferId);
+ assertThat(retrieved).isEqualTo(transfer);
+ }
+
+ @Test
+ public void testReceivedTransfer()
+ {
+ TimeUUID planId = nextTimeUUID();
+ PendingLocalTransfer transfer = pendingTransfer(planId);
+ localTransfers.received(transfer);
+ PendingLocalTransfer retrieved = localTransfers.getPendingTransfer(planId);
+ assertThat(retrieved).isEqualTo(transfer);
+ }
+
+ @Test
+ public void testReceivedEmptyTransferThrows()
+ {
+ assertThatThrownBy(() -> new PendingLocalTransfer(planId, Collections.emptyList()))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ public void testGetPendingTransferNotFound()
+ {
+ PendingLocalTransfer retrieved = localTransfers.getPendingTransfer(planId);
+ assertThat(retrieved).isNull();
+ }
+
+ @Test
+ public void testGetActivatedTransferNotFound()
+ {
+ CoordinatedTransfer retrieved = localTransfers.getActivatedTransfer(transferId);
+ assertThat(retrieved).isNull();
+ }
+
+ @Test
+ public void testPurgingTransferNotStarted()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // All streams in INIT state - should NOT be purgeable (stream hasn't started yet)
+ CoordinatedTransfer.SingleTransferResult result = CoordinatedTransfer.SingleTransferResult.Init();
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isFalse();
+ }
+
+ @Test
+ public void testPurgingTransferAllStreamsComplete()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // All streams in STREAM_COMPLETE state - should NOT be purgeable (no failures)
+ CoordinatedTransfer.SingleTransferResult result1 = CoordinatedTransfer.SingleTransferResult.StreamComplete(nextTimeUUID());
+ CoordinatedTransfer.SingleTransferResult result2 = CoordinatedTransfer.SingleTransferResult.StreamComplete(nextTimeUUID());
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isFalse();
+ }
+
+ @Test
+ public void testPurgingTransferPrepareFailed()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(PREPARE_FAILED, planId);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(PREPARING, planId);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isTrue();
+ }
+
+ @Test
+ public void testPurgingTransferAllActivationCommitted()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // All streams in ACTIVATE_COMMITTED state - should be purgeable (allComplete = true)
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(COMMITTED, planId);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(COMMITTED, planId);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isTrue();
+ }
+
+ @Test
+ public void testPurgingTransferMixedCommittedAndNoop()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // Mix of ACTIVATE_COMMITTED and STREAM_NOOP - should be purgeable (allComplete = true)
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(COMMITTED, planId);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(STREAM_NOOP, null);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isTrue();
+ }
+
+ @Test
+ public void testPurgingTransferActivationPartialCommitted()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // One stream in ACTIVATE_PREPARING - should NOT be purgeable
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(PREPARING, planId);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(COMMITTING, planId);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isFalse();
+ }
+
+ @Test
+ public void testPurgingTransferAllStreamsFailed()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // All streams in STREAM_FAILED state - should be purgeable (noneActivated = true)
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(STREAM_FAILED, planId);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(STREAM_FAILED, planId);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isTrue();
+ }
+
+ @Test
+ public void testPurgingTransferMixedInitAndFailed()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // Mix of INIT and STREAM_FAILED - should be purgeable (has failure, none activated)
+ CoordinatedTransfer.SingleTransferResult result1 = CoordinatedTransfer.SingleTransferResult.Init();
+ CoordinatedTransfer.SingleTransferResult result2 = CoordinatedTransfer.SingleTransferResult.Init().streamFailed(nextTimeUUID());
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isTrue();
+ }
+
+ @Test
+ public void testPurgingTransferMixedCompleteAndFailed()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // Mix of STREAM_COMPLETE and STREAM_FAILED - should be purgeable (has failure, none activated)
+ CoordinatedTransfer.SingleTransferResult result1 = CoordinatedTransfer.SingleTransferResult.StreamComplete(nextTimeUUID());
+ CoordinatedTransfer.SingleTransferResult result2 = CoordinatedTransfer.SingleTransferResult.Init().streamFailed(nextTimeUUID());
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isTrue();
+ }
+
+ @Test
+ public void testPurgingTransferMixedStreamingCompleteAndPreparing()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // Mix of STREAM_COMPLETE and ACTIVATE_PREPARING - should NOT be purgeable
+ // (noneActivated = false because of ACTIVATE_PREPARING, allComplete = false)
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(STREAM_COMPLETE, planId);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(PREPARING, planId);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isFalse();
+ }
+
+ @Test
+ public void testPurgingTransferMixedCommittingCommitted()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(COMMITTING, planId);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(COMMITTED, planId);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isFalse();
+ }
+
+ @Test
+ public void testPurgingTransferWithNullTransferId()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(null);
+
+ // All streams complete but transferId is null - should NOT be purgeable
+ CoordinatedTransfer.SingleTransferResult result1 = new CoordinatedTransfer.SingleTransferResult(STREAM_COMPLETE, null);
+ CoordinatedTransfer.SingleTransferResult result2 = new CoordinatedTransfer.SingleTransferResult(STREAM_COMPLETE, null);
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ // allComplete = true, but transferId is null, so should not purge
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isFalse();
+ }
+
+ @Test
+ public void testPurgingTransferNoopOnly()
+ {
+ CoordinatedTransfer transfer = coordinatedTransfer(transferId);
+
+ // All streams in STREAM_NOOP - should be purgeable (both noneActivated and allComplete are true)
+ CoordinatedTransfer.SingleTransferResult result1 = CoordinatedTransfer.SingleTransferResult.Noop();
+ CoordinatedTransfer.SingleTransferResult result2 = CoordinatedTransfer.SingleTransferResult.Noop();
+
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result1);
+ transfer.streamResults.put(mock(InetAddressAndPort.class), result2);
+
+ Assertions.assertThat(localTransfers.purger.test(transfer)).isTrue();
+ }
+}
diff --git a/test/unit/org/apache/cassandra/replication/TransferActivationSerializationTest.java b/test/unit/org/apache/cassandra/replication/TransferActivationSerializationTest.java
new file mode 100644
index 0000000000..57a51f87b8
--- /dev/null
+++ b/test/unit/org/apache/cassandra/replication/TransferActivationSerializationTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.Serializers;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
+
+public class TransferActivationSerializationTest
+{
+ private static final int VERSION = MessagingService.current_version;
+
+ @BeforeClass
+ public static void setUpClass()
+ {
+ DatabaseDescriptor.daemonInitialization();
+ }
+
+ @Test
+ public void testRoundtripPreparePhase() throws IOException
+ {
+ TimeUUID planId = nextTimeUUID();
+ MutationId transferId = new MutationId(1L, 100, 200);
+ NodeId coordinatorId = new NodeId(1);
+ TransferActivation.Phase phase = TransferActivation.Phase.PREPARE;
+
+ TransferActivation activation = new TransferActivation(planId, transferId, coordinatorId, phase);
+
+ try (DataOutputBuffer output = new DataOutputBuffer())
+ {
+ Serializers.testSerde(output, TransferActivation.serializer, activation, VERSION);
+ }
+ }
+
+ @Test
+ public void testRoundtripCommitPhase() throws IOException
+ {
+ TimeUUID planId = nextTimeUUID();
+ MutationId transferId = new MutationId(2L, 300, 400);
+ NodeId coordinatorId = new NodeId(2);
+ TransferActivation.Phase phase = TransferActivation.Phase.COMMIT;
+
+ TransferActivation activation = new TransferActivation(planId, transferId, coordinatorId, phase);
+
+ try (DataOutputBuffer output = new DataOutputBuffer())
+ {
+ Serializers.testSerde(output, TransferActivation.serializer, activation, VERSION);
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/tcm/ownership/ReplicaGroupsTest.java b/test/unit/org/apache/cassandra/tcm/ownership/ReplicaGroupsTest.java
new file mode 100644
index 0000000000..5efbb80f55
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tcm/ownership/ReplicaGroupsTest.java
@@ -0,0 +1,228 @@
+/*
+ * 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.tcm.ownership;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import accord.utils.Gen;
+import accord.utils.Gens;
+import accord.utils.RandomSource;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.RandomPartitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Splitter;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.locator.EndpointsForRange;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.utils.AccordGenerators;
+import org.apache.cassandra.utils.CassandraGenerators;
+import org.assertj.core.api.Assertions;
+
+import static accord.utils.Property.qt;
+import static org.apache.cassandra.tcm.membership.MembershipUtils.endpoint;
+
+@RunWith(Parameterized.class)
+public class ReplicaGroupsTest
+{
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<Object[]> data()
+ {
+ return Arrays.asList(new Object[][] {
+ { RandomPartitioner.instance },
+ { Murmur3Partitioner.instance },
+ });
+ }
+
+ private final IPartitioner partitioner;
+
+ public ReplicaGroupsTest(IPartitioner partitioner)
+ {
+ this.partitioner = partitioner;
+ }
+
+ @Test
+ public void testMatchToken()
+ {
+ qt().forAll(args(partitioner))
+ .check(args -> {
+ for (Token token : args.tokens)
+ assertEquivalent(args.groups, token);
+ });
+ }
+
+ @Test
+ public void minimumRepro()
+ {
+ Range<Token> range = new Range<>(new Murmur3Partitioner.LongToken(5), new Murmur3Partitioner.LongToken(10));
+ EndpointsForRange.Builder endpointsBuilder = EndpointsForRange.builder(range);
+ endpointsBuilder.add(Replica.fullReplica(endpoint((byte) 1), range));
+
+ ReplicaGroups groups = ReplicaGroups.builder()
+ .withReplicaGroup(VersionedEndpoints.forRange(Epoch.FIRST, endpointsBuilder.build()))
+ .build();
+
+ Token token = new Murmur3Partitioner.LongToken(5);
+ assertEquivalent(groups, token);
+ }
+
+ @Test
+ public void testMatchTokenWithWraparoundRange()
+ {
+ // Create a wraparound range (wraps around from high to low token value)
+ // For RandomPartitioner, this would be something like (100, minToken] which wraps around
+ Token highToken = new RandomPartitioner.BigIntegerToken("100");
+ Token minToken = new RandomPartitioner.BigIntegerToken("0");
+ Range<Token> wraparoundRange = new Range<>(highToken, minToken);
+
+ EndpointsForRange.Builder endpointsBuilder = EndpointsForRange.builder(wraparoundRange);
+ endpointsBuilder.add(Replica.fullReplica(endpoint((byte) 1), wraparoundRange));
+
+ ReplicaGroups groups = ReplicaGroups.builder()
+ .withReplicaGroup(VersionedEndpoints.forRange(Epoch.FIRST, endpointsBuilder.build()))
+ .build();
+
+ // Test with minToken - should be in the wraparound range
+ assertEquivalent(groups, minToken);
+
+ // Test with a token just after highToken
+ Token tokenAfterHigh = new RandomPartitioner.BigIntegerToken("150");
+ assertEquivalent(groups, tokenAfterHigh);
+ }
+
+ private static Object result(Callable<Object> callable)
+ {
+ try
+ {
+ return callable.call();
+ }
+ catch (Throwable t)
+ {
+ return t;
+ }
+ }
+
+ private static void assertEquivalent(ReplicaGroups groups, Token token)
+ {
+ Object fast = result(() -> groups.matchToken(token));
+ Object slow = result(() -> {
+ for (VersionedEndpoints.ForRange forRange : groups.endpoints)
+ {
+ if (forRange.get().range().contains(token))
+ return forRange;
+ }
+ return null;
+ });
+ if (fast instanceof Throwable && slow instanceof Throwable)
+ {
+ Assertions.assertThat(fast).hasSameClassAs(slow);
+ Assertions.assertThat(((Throwable) fast).getMessage()).isEqualTo(((Throwable) slow).getMessage());
+ }
+ else
+ {
+ Assertions.assertThat(slow).isEqualTo(fast);
+ }
+ }
+
+ private static class Args
+ {
+ ReplicaGroups groups;
+ Collection<Token> tokens;
+
+ public Args(ReplicaGroups groups, Collection<Token> tokens)
+ {
+ this.groups = groups;
+ this.tokens = tokens;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Args{" +
+ "groups=" + groups +
+ ", tokens=" + tokens +
+ '}';
+ }
+ }
+
+ private static Gen<Args> args(IPartitioner partitioner)
+ {
+ return rs -> new Args(replicaGroups(rs, partitioner), tokens(rs, partitioner));
+ }
+
+ private static ReplicaGroups replicaGroups(RandomSource rs, IPartitioner partitioner)
+ {
+ Range<Token> full = new Range<>(partitioner.getMinimumToken(), partitioner.getMaximumTokenForSplitting());
+ Splitter splitter = partitioner.splitter().get();
+ int parts = rs.nextBiasedInt(0, 5, 100);
+ Set<Range<Token>> splits = splitter.split(full, parts);
+
+ // Drop some ranges so there are gaps
+ List<Token> drops = Gens.lists(AccordGenerators.fromQT(CassandraGenerators.tokensInRange(full))).ofSizeBetween(0, 2).next(rs);
+
+ ReplicaGroups.Builder builder = ReplicaGroups.builder();
+
+ for (Range<Token> range : splits)
+ {
+ boolean skip = false;
+ for (Token drop : drops)
+ if (range.contains(drop))
+ skip = true;
+ if (skip)
+ continue;
+
+ // Generate 1-3 replicas per range
+ int numReplicas = rs.nextInt(1, 3);
+ EndpointsForRange.Builder endpointsBuilder = EndpointsForRange.builder(range);
+
+ for (int i = 0; i < numReplicas; i++)
+ {
+ InetAddressAndPort endpoint = endpoint((byte) (i + 1));
+ endpointsBuilder.add(Replica.fullReplica(endpoint, range));
+ }
+
+ builder.withReplicaGroup(VersionedEndpoints.forRange(Epoch.FIRST, endpointsBuilder.build()));
+ }
+
+ return builder.build();
+ }
+
+ private static Collection<Token> tokens(RandomSource rs, IPartitioner partitioner)
+ {
+ Gen<Token> tokenGen = AccordGenerators.fromQT(CassandraGenerators.token(partitioner));
+ int numTokens = rs.nextInt(1, 3);
+ List<Token> tokens = new ArrayList<>();
+ for (int i = 0; i < numTokens; i++)
+ {
+ tokens.add(tokenGen.next(rs));
+ }
+ return tokens;
+ }
+}
diff --git a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
index 71b18c2905..444c93d01e 100644
--- a/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
+++ b/test/unit/org/apache/cassandra/utils/CassandraGenerators.java
@@ -1442,6 +1442,50 @@ public final class CassandraGenerators
}
}
+ public static Gen<Token> randomTokenIn(Range<Token> range)
+ {
+ BigInteger left = (BigInteger) range.left.getTokenValue();
+ BigInteger right = (BigInteger) range.right.getTokenValue();
+ BigInteger min = RandomPartitioner.instance.getMinimumToken().getTokenValue();
+ BigInteger max = (BigInteger) RandomPartitioner.instance.getMaximumTokenForSplitting().getTokenValue();
+ Gen<byte[]> bytesGen = Generate.byteArrays(SourceDSL.integers().between(0, RandomPartitioner.MAXIMUM_TOKEN_SIZE), Generate.bytes(Byte.MIN_VALUE, Byte.MAX_VALUE, (byte) 0));
+
+ return rs -> {
+ BigInteger from;
+ BigInteger to;
+
+ if (range.isWrapAround())
+ {
+ Boolean inUpperRange = SourceDSL.booleans().all().generate(rs);
+ if (inUpperRange)
+ {
+ from = right;
+ to = max;
+ }
+ else
+ {
+ from = min;
+ to = left;
+ }
+ }
+ else
+ {
+ from = left;
+ to = right;
+ }
+
+ byte[] bytes = bytesGen.generate(rs);
+ BigInteger random = bytes.length == 0 ? min : new BigInteger(bytes);
+ BigInteger delta = to.subtract(from);
+ BigInteger mod;
+ if (delta.equals(BigInteger.ZERO))
+ mod = min;
+ else
+ mod = random.mod(delta);
+ return RandomPartitioner.instance.getToken(ByteBuffer.wrap(mod.toByteArray()));
+ };
+ }
+
public static Gen<Token> byteOrderToken()
{
// empty token only happens if partition key is byte[0], which isn't allowed
@@ -1510,6 +1554,7 @@ public final class CassandraGenerators
{
IPartitioner partitioner = range.left.getPartitioner();
if (partitioner instanceof Murmur3Partitioner) return murmurTokenIn(range);
+ if (partitioner instanceof RandomPartitioner) return randomTokenIn(range);
throw new UnsupportedOperationException("Unsupported partitioner: " + partitioner.getClass());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org