This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-6.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-6.0 by this push:
new 03422edb4b Relax assertion on partitioner instances in
SinglePartitionReadCommand
03422edb4b is described below
commit 03422edb4bc13c34dffca5cce2e29603db436513
Author: Sam Tunnicliffe <[email protected]>
AuthorDate: Wed Jun 24 11:23:39 2026 +0100
Relax assertion on partitioner instances in SinglePartitionReadCommand
Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-21251
---
CHANGES.txt | 1 +
.../cassandra/db/SinglePartitionReadCommand.java | 4 +-
.../org/apache/cassandra/dht/IPartitioner.java | 9 ++
...ndexTablePartitionerAfterForceSnapshotTest.java | 141 +++++++++++++++++++++
.../upgrade/ClusterMetadata2iInitializeTest.java | 112 ++++++++++++++++
5 files changed, 266 insertions(+), 1 deletion(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 36dd204c0f..eef9e5b576 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
6.0-alpha2
+ * Relax assertion on partitioner instances in SinglePartitionReadCommand
(CASSANDRA-21251)
* Report cancelled read command execution to coordinator as a
RequestFailure.TIMEOUT (CASSANDRA-21468)
* Fix TCM log catchup from peer with snapshots and gaps in the log sequence
(CASSANDRA-21455)
* Speed up nodetool doc generation by producing all command help in a single
jvm (CASSANDRA-21444)
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index 28ae9b786a..567d9f066c 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@ -70,6 +70,7 @@ import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry;
import org.apache.cassandra.db.virtual.VirtualTable;
import org.apache.cassandra.dht.Bounds;
+import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.io.sstable.SSTableReadsListener;
@@ -121,7 +122,8 @@ public class SinglePartitionReadCommand extends ReadCommand
implements SinglePar
DataRange dataRange)
{
super(serializedAtEpoch, Kind.SINGLE_PARTITION, isDigest,
digestVersion, acceptsTransient, potentialTxnConflicts, metadata, nowInSec,
columnFilter, rowFilter, limits, indexQueryPlan, trackWarnings, dataRange);
- assert partitionKey.getPartitioner() == metadata.partitioner;
+ assert IPartitioner.equivalent(partitionKey.getPartitioner(),
metadata.partitioner) : String.format("Mismatching partitioners for key (%s)
and table metadata (%s)",
+
partitionKey.getPartitioner(),
metadata.partitioner);
this.partitionKey = partitionKey;
this.clusteringIndexFilter = clusteringIndexFilter;
}
diff --git a/src/java/org/apache/cassandra/dht/IPartitioner.java
b/src/java/org/apache/cassandra/dht/IPartitioner.java
index 6adc9d92fa..dfd26f6165 100644
--- a/src/java/org/apache/cassandra/dht/IPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/IPartitioner.java
@@ -41,6 +41,15 @@ public interface IPartitioner
return DatabaseDescriptor.getPartitioner();
}
+ static boolean equivalent(IPartitioner p1, IPartitioner p2)
+ {
+ if (p1 == p2) return true;
+ if (p1.getClass() != p2.getClass()) return false;
+ if (p1.getClass() == LocalPartitioner.class)
+ return ((LocalPartitioner)p1).comparator ==
((LocalPartitioner)p2).comparator;
+ return true;
+ }
+
/**
* @return a new instance of a reusable key
*/
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tcm/IndexTablePartitionerAfterForceSnapshotTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tcm/IndexTablePartitionerAfterForceSnapshotTest.java
new file mode 100644
index 0000000000..ed0cbfd46d
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/tcm/IndexTablePartitionerAfterForceSnapshotTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.tcm;
+
+import java.io.IOException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
+import org.awaitility.Awaitility;
+import org.junit.Test;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.rows.UnfilteredRowIterator;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.index.internal.CassandraIndexSearcher;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.tcm.Epoch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.junit.Assert.assertEquals;
+
+public class IndexTablePartitionerAfterForceSnapshotTest extends TestBaseImpl
+{
+ @Test
+ public void indexQueryAfterSnapshotTest() throws IOException,
InterruptedException
+ {
+ try (Cluster cluster = Cluster.build(3)
+
.withInstanceInitializer(BBInstaller::install)
+ .withConfig(config ->
config.with(GOSSIP, NETWORK))
+ .start())
+ {
+ init(cluster);
+ cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int,
ck int, v int, PRIMARY KEY (pk, ck))");
+ cluster.schemaChange(withKeyspace("create index iii2 on %s.tbl
(v)"));
+ for (int i = 0; i < 10000; i++)
+ cluster.coordinator(1).execute(withKeyspace("insert into
%s.tbl (pk, ck, v) values (?, ?, ?)"), ConsistencyLevel.ALL, 1, i, i % 5);
+
+ // drop metadata change replication messages from node 1 to nodes
2 & 3
+ cluster.filters().verbs(Verb.TCM_REPLICATION.id,
+ Verb.TCM_FETCH_CMS_LOG_RSP.id,
+ Verb.TCM_FETCH_PEER_LOG_RSP.id,
+ Verb.TCM_CURRENT_EPOCH_REQ.id)
+ .from(1)
+ .drop()
+ .on();
+
+ // node1 makes some metadata changes interspersed with snapshots.
When the message filters are dropped, this
+ // will cause nodes 2 & 3 to try and catchup. We want multiple
snapshots to be taken to ensure that the
+ // catchup responses contain one, rather than just a list of
entries.
+ cluster.get(1).nodetoolResult("cms",
"snapshot").asserts().success();
+ cluster.get(1).nodetoolResult("cms",
"snapshot").asserts().success();
+ final Epoch epoch = ClusterUtils.getCurrentEpoch(cluster.get(1));
+ // start executing
+ AtomicBoolean stop = new AtomicBoolean();
+ AtomicInteger queryFailures = new AtomicInteger();
+ Thread t = new Thread(() -> {
+ while (!stop.get())
+ {
+ try
+ {
+ cluster.coordinator(1).execute(withKeyspace("select *
from %s.tbl where pk=1 and v=4"), ConsistencyLevel.ALL);
+ }
+ catch (Throwable e)
+ {
+ queryFailures.incrementAndGet();
+ }
+ }
+ });
+ t.start();
+ TimeUnit.SECONDS.sleep(1);
+ // drop the filters and wait for nodes 2 & 3 to catch up
+ cluster.filters().reset();
+ Awaitility.waitAtMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until(() ->
ClusterUtils.getCurrentEpoch(cluster.get(2)).isEqualOrAfter(epoch) &&
+
ClusterUtils.getCurrentEpoch(cluster.get(3)).isEqualOrAfter(epoch));
+
+ // give it another second of querying
+ TimeUnit.SECONDS.sleep(1);
+ stop.set(true);
+ t.join();
+ assertEquals(0, queryFailures.get());
+ }
+ }
+
+ public static class BBInstaller
+ {
+ public static void install(ClassLoader classLoader, int inst)
+ {
+ if (inst == 1)
+ return;
+ new ByteBuddy().rebase(CassandraIndexSearcher.class)
+ .method(named("queryIndex"))
+ .intercept(MethodDelegation.to(BBInterceptor.class))
+ .make()
+ .load(classLoader,
ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+
+ public static class BBInterceptor
+ {
+ public static UnfilteredRowIterator queryIndex(DecoratedKey indexKey,
ReadCommand command, ReadExecutionController executionController, @SuperCall
Callable<UnfilteredRowIterator> zuper) throws Exception
+ {
+ // this makes it more likely that we decorate the key with one
partitioner and execute the SinglePartitionReadCommand with another
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ return zuper.call();
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadata2iInitializeTest.java
b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadata2iInitializeTest.java
new file mode 100644
index 0000000000..8d4ee9f480
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadata2iInitializeTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.upgrade;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.rows.UnfilteredRowIterator;
+import org.apache.cassandra.distributed.UpgradeableCluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.index.internal.CassandraIndexSearcher;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.junit.Assert.assertEquals;
+
+public class ClusterMetadata2iInitializeTest extends UpgradeTestBase
+{
+ @Test
+ public void initializeCMSWithConcurrentIndexReadsTest() throws Throwable
+ {
+ Consumer<UpgradeableCluster.Builder > builderUpdater = builder ->
builder.withInstanceInitializer(BBInstaller::install);
+ new TestCase()
+ .nodes(3)
+ .withConfig((cfg) -> cfg.with(Feature.GOSSIP))
+ .withBuilder(builderUpdater)
+ .upgradesToCurrentFrom(v41)
+ .setup((cluster) -> {
+ cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int,
ck int, v int, PRIMARY KEY (pk, ck))");
+ cluster.schemaChange(withKeyspace("create index iii2 on %s.tbl
(v)"));
+ for (int i = 0; i < 10000; i++)
+ cluster.coordinator(1).execute(withKeyspace("insert into
%s.tbl (pk, ck, v) values (?, ?, ?)"), ConsistencyLevel.ALL, 1, i, i%5);
+ })
+ .runAfterClusterUpgrade((cluster) -> {
+ AtomicBoolean stop = new AtomicBoolean();
+ AtomicInteger queryFailures = new AtomicInteger();
+ Thread t = new Thread(() -> {
+ while (!stop.get())
+ {
+ try
+ {
+ cluster.coordinator(1).execute(withKeyspace("select *
from %s.tbl where pk=1 and v=4"), ConsistencyLevel.ALL);
+ }
+ catch (Throwable e)
+ {
+ queryFailures.incrementAndGet();
+ }
+ }
+ });
+ t.start();
+ cluster.get(1).nodetoolResult("cms",
"initialize").asserts().success();
+ stop.set(true);
+ t.join();
+ assertEquals(0, queryFailures.get());
+ }).run();
+ }
+
+ public static class BBInstaller
+ {
+ public static void install(ClassLoader classLoader, int inst)
+ {
+ if (inst == 1)
+ return;
+ new ByteBuddy().rebase(CassandraIndexSearcher.class)
+ .method(named("queryIndex"))
+ .intercept(MethodDelegation.to(BBInterceptor.class))
+ .make()
+ .load(classLoader,
ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+
+ public static class BBInterceptor
+ {
+ public static UnfilteredRowIterator queryIndex(DecoratedKey indexKey,
ReadCommand command, ReadExecutionController executionController, @SuperCall
Callable<UnfilteredRowIterator> zuper) throws Exception
+ {
+ // this makes it more likely that we decorate the key with one
partitioner and execute the SinglePartitionReadCommand with another
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ return zuper.call();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
