This is an automated email from the ASF dual-hosted git repository.
sarankk pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new 330134e231 CASSANDRA-21723 Wrap-around shard silently skipped during
coordinated SSTable import for mutation tracking enabled tables
330134e231 is described below
commit 330134e2312c8ab60740c3a84d0af3cc709edcbd
Author: Saranya Krishnakumar <[email protected]>
AuthorDate: Tue Oct 20 05:58:59 2026 -0700
CASSANDRA-21723 Wrap-around shard silently skipped during coordinated
SSTable import for mutation tracking enabled tables
---
CHANGES.txt | 1 +
.../replication/TrackedImportTransfers.java | 4 +-
.../test/tracking/TrackedImportTransferTest.java | 58 ++++++++++++++++++++++
3 files changed, 61 insertions(+), 2 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 55ef96c0a0..f15927c962 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
cep-45-mutation-tracking
+ * CEP-45: Wrap-around shard silently skipped during coordinated SSTable
import for tracked tables (CASSANDRA-21723)
* CEP-45: Migration repair streams can write live SSTables into
pending/<planId>/ and lose them silently on restart (CASSANDRA-21626)
* CEP-45: Prevent dropping journal segments referenced by partially
reconciled sstables (CASSANDRA-21406)
* CEP-45: Fixes and improvements to mutation tracking migration
(CASSANDSA-21653)
diff --git
a/src/java/org/apache/cassandra/replication/TrackedImportTransfers.java
b/src/java/org/apache/cassandra/replication/TrackedImportTransfers.java
index b2974df4c6..364ae13690 100644
--- a/src/java/org/apache/cassandra/replication/TrackedImportTransfers.java
+++ b/src/java/org/apache/cassandra/replication/TrackedImportTransfers.java
@@ -30,10 +30,10 @@ import java.util.Map;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
+import org.apache.cassandra.db.lifecycle.View;
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 TrackedImportTransfer}
instances from a collection
@@ -69,7 +69,7 @@ public class TrackedImportTransfers implements
Iterable<TrackedImportTransfer>
shards.forEachShard(shard -> {
Range<Token> range = shard.tokenRange();
- Collection<SSTableReader> sstablesForRange =
intervals.search(Interval.create(range.left.minKeyBound(),
range.right.maxKeyBound()));
+ Collection<SSTableReader> sstablesForRange =
View.sstablesInBounds(range.left.minKeyBound(), range.right.maxKeyBound(),
intervals);
List<Range<Token>> ranges = Collections.singletonList(range);
Map<SSTableReader, List<SSTableReader.PartitionPositionBounds>>
positionForSSTables = new HashMap<>();
sstablesForRange.removeIf(sstable -> {
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportTransferTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportTransferTest.java
index 9095d82db3..602d980f4b 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportTransferTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportTransferTest.java
@@ -19,8 +19,10 @@
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.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -38,6 +40,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Range;
+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.IInvokableInstance;
@@ -45,6 +50,7 @@ import org.apache.cassandra.distributed.shared.AssertUtils;
import org.apache.cassandra.distributed.test.sai.SAIUtil;
import org.apache.cassandra.io.sstable.CQLSSTableWriter;
import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ownership.DataPlacement;
@@ -247,6 +253,58 @@ public class TrackedImportTransferTest extends
TrackedTransferTestBase
assertLocalSelect(cluster, keyspace, rows -> assertRows(rows,
row(KEY_100, 1), row(KEY_300, 1)));
}
+ @Test
+ public void importWrapAroundShard() throws IOException
+ {
+ // See CASSANDRA-21723
+ String keyspace = "wrap_around_shard";
+ cluster.schemaChange("CREATE KEYSPACE " + keyspace + " WITH
replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND
replication_type='tracked';");
+ String createTableQuery = "CREATE TABLE " +
tableWithKeyspace(keyspace) + " (k BLOB PRIMARY KEY, v INT)";
+ cluster.schemaChange(createTableQuery);
+
+ // find wrap around key for write
+ long wrapAroundLeft = cluster.get(1).callOnInstance(() -> {
+ List<Range<Token>> wrapping = new ArrayList<>();
+
MutationTrackingService.instance().forEachShardInKeyspace(keyspace, shard -> {
+ Range<Token> range = shard.tokenRange();
+ if (range.right.isMinimum())
+ wrapping.add(range);
+ });
+ assertThat(wrapping).hasSize(1);
+ return wrapping.get(0).left.getLongValue();
+ });
+ ByteBuffer wrapAroundKey =
Murmur3Partitioner.LongToken.keyForToken(wrapAroundLeft + 1);
+
+ String file =
Files.createTempDirectory(TrackedTransferTestBase.class.getSimpleName()).toString();
+ CQLSSTableWriter.Builder builder = CQLSSTableWriter.builder()
+
.forTable(createTableQuery)
+ .inDirectory(file)
+ .using("INSERT INTO
" + tableWithKeyspace(keyspace) + " (k, v) " + "VALUES (?, ?)");
+
+ // One row per shard
+ try (CQLSSTableWriter writer = builder.build())
+ {
+ writer.addRow(KEY_100, 1); // -4074457345618258601L
+ writer.addRow(KEY_200, 1); // 1L
+ writer.addRow(KEY_300, 1); // 3074457345618258602L
+ writer.addRow(wrapAroundKey, 1); // wrapAroundLeft + 1
+ }
+
+ assertLocalSelect(cluster, keyspace, AssertUtils::assertRows);
+ List<String> failed = cluster.get(1).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 import
+ Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
+
+ Assertions.assertThat(failed).isEmpty();
+ assertLocalSelect(cluster, keyspace, rows -> assertRows(rows,
row(KEY_100, 1), row(KEY_200, 1), row(KEY_300, 1), row(wrapAroundKey, 1)));
+ }
+
@Test
public void importMoreThanOneSSTable() throws IOException
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
