You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sa...@apache.org on 2025/11/25 17:53:32 UTC
(cassandra) branch trunk updated (2acc420e1c -> 209288cebe)
This is an automated email from the ASF dual-hosted git repository.
samt pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
from 2acc420e1c Avoid iterating all prepared statements when getting PreparedStatementsCacheSize metric
new 44f6a32718 AlterSchemaStatement with no effect locally isn't submitted to CMS
new 856db2afa0 Retry without time limit calculates wait time correctly
new 62bd9ab1ae Gossip entries for hibernating non-members don't block truncate
new ec7794f20f Avoid NPE when meta keyspace placements are empty before CMS is initialized
new 496959a8f2 Don't update registration status if node state for decommissioned peer is found with the same address
new 209288cebe Reject PrepareJoin if tokens already assigned
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 6 +
.../statements/schema/AlterSchemaStatement.java | 10 +-
src/java/org/apache/cassandra/gms/Gossiper.java | 5 +
.../apache/cassandra/service/CassandraDaemon.java | 8 +-
.../apache/cassandra/service/StorageService.java | 8 +-
.../org/apache/cassandra/tcm/ClusterMetadata.java | 5 +-
src/java/org/apache/cassandra/tcm/Processor.java | 36 ++---
src/java/org/apache/cassandra/tcm/Retry.java | 2 +-
.../apache/cassandra/tcm/membership/Directory.java | 3 +-
.../cassandra/tcm/transformations/PrepareJoin.java | 13 ++
.../distributed/impl/AbstractCluster.java | 11 ++
.../test/HibernatingGossipOnlyMembersTest.java | 67 ++++++++
.../distributed/test/ring/DecommissionTest.java | 49 ++++++
.../ClusterMetadataSingleNodeUpgradeTest.java | 5 +
.../cql3/AlterSchemaStatementNoOpTest.java | 173 +++++++++++++++++++++
.../cassandra/cql3/PreparedStatementsTest.java | 91 +++++------
.../service/accord/AccordTopologyUtils.java | 2 +-
.../reads/repair/AbstractReadRepairTest.java | 6 +-
test/unit/org/apache/cassandra/tcm/RetryTest.java | 111 +++++++++++++
.../tcm/transformations/PrepareJoinTest.java | 130 ++++++++++++++++
.../tcm/transformations/PrepareLeaveTest.java | 26 +++-
21 files changed, 669 insertions(+), 98 deletions(-)
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/HibernatingGossipOnlyMembersTest.java
create mode 100644 test/unit/org/apache/cassandra/cql3/AlterSchemaStatementNoOpTest.java
create mode 100644 test/unit/org/apache/cassandra/tcm/RetryTest.java
create mode 100644 test/unit/org/apache/cassandra/tcm/transformations/PrepareJoinTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 05/06: Don't update registration status if node state for decommissioned peer is found with the same address
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 496959a8f21adbcdfee0a0a586d9d2596a5d856f
Author: Sam Tunnicliffe <sa...@apache.org>
AuthorDate: Mon Nov 10 09:09:56 2025 +0000
Don't update registration status if node state for decommissioned peer is found with the same address
Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-21005
---
CHANGES.txt | 1 +
.../apache/cassandra/service/CassandraDaemon.java | 8 +++-
.../distributed/impl/AbstractCluster.java | 11 +++++
.../distributed/test/ring/DecommissionTest.java | 49 ++++++++++++++++++++++
4 files changed, 68 insertions(+), 1 deletion(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index ac61998c3a..7b7ff72e81 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.1
+ * Don't update registration status if node state for decommissioned peer is found with the same address (CASSANDRA-21005)
* Avoid NPE when meta keyspace placements are empty before CMS is initialized (CASSANDRA-21004)
* Gossip entries for hibernating non-members don't block truncate (CASSANDRA-21003)
* Retry without time limit calculates wait time incorrectly (CASSANDRA-21002)
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 2f85e6e58e..973dfbfe92 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -90,6 +90,7 @@ import org.apache.cassandra.service.snapshot.SnapshotManager;
import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.MultiStepOperation;
+import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JMXServerUtils;
import org.apache.cassandra.utils.JVMStabilityInspector;
@@ -278,7 +279,12 @@ public class CassandraDaemon
disableAutoCompaction(Schema.instance.distributedKeyspaces().names());
CMSOperations.initJmx();
AccordOperations.initJmx();
- if (ClusterMetadata.current().myNodeId() != null)
+ NodeState nodeStateForLocalAddress = ClusterMetadata.current().myNodeState();
+ // If another node with the same address was previously a member and was decommissioned, it can be
+ // present in ClusterMetadata with a LEFT state. That should not trigger _this_ node to update
+ // RegistrationStatus. During the startup process the old node will be expunged and this node
+ // will register, prompting another call to onRegistration.
+ if (nodeStateForLocalAddress != null && nodeStateForLocalAddress != NodeState.LEFT)
RegistrationStatus.instance.onRegistration();
}
catch (InterruptedException | ExecutionException | IOException e)
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index 224adabdb1..533a7b49d5 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@ -697,6 +697,17 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
return instance;
}
+ public synchronized void unsafeRemoveNode(I toRemove)
+ {
+ instances.remove(toRemove);
+ instanceMap.remove(toRemove.broadcastAddress(), toRemove);
+ }
+
+ public void unsafeUpdateNodeIdTopology(int num, NetworkTopology.DcAndRack location)
+ {
+ nodeIdTopology.put(num, location);
+ }
+
/**
* WARNING: we index from 1 here, for consistency with inet address!
*/
diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
index 6f67dac03b..db2f2eca6d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
@@ -34,13 +34,19 @@ import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.io.util.File;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.tcm.ClusterMetadata;
@@ -55,6 +61,8 @@ import org.apache.cassandra.utils.FBUtilities;
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.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack;
+import static org.apache.cassandra.distributed.shared.NetworkTopology.networkTopology;
import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -75,6 +83,47 @@ public class DecommissionTest extends TestBaseImpl
}
}
+ @Test
+ public void testAddressReuseAfterDecommission() throws IOException, ExecutionException, InterruptedException
+ {
+ // Initially, all nodes should be in dc1/rack1. Node 3 will be decommissioned and a new node added re-using
+ // node 3's address. When the new node registers, it should be in dc2/rack2.
+ // For now, this requires the accord service to disabled. See CASSANDRA-21026
+ try (Cluster cluster = builder().withNodes(3)
+ .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4))
+ .withConfig(config -> config.with(NETWORK, GOSSIP)
+ .set("accord.enabled", false))
+ .withNodeIdTopology(networkTopology(3, (id) -> dcAndRack("dc1", "rack1")))
+ .start())
+ {
+ assertEquals("dc1/rack1", cluster.get(1).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc1/rack1", cluster.get(2).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc1/rack1", cluster.get(3).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+
+ IInvokableInstance toRemove = cluster.get(3);
+ toRemove.nodetoolResult("decommission", "--force").asserts().success();
+ toRemove.shutdown().get();
+ ClusterUtils.getDirectories(toRemove).forEach(File::tryDeleteRecursive);
+ cluster.unsafeRemoveNode(toRemove);
+
+ // Now add a new node, using the same address as the one we just removed. This new node should register
+ // itself in dc2/rack2 and not inherit the location of its predecessor.
+ // Note: because we have removed the original node3 from the cluster completely, which is necessary because
+ // the cluster will complain about an id clash otherwise, this new node will also be "node3". However, it is
+ // completely distinct from the original one.
+ cluster.unsafeUpdateNodeIdTopology(toRemove.config().num(), dcAndRack("dc2", "rack2"));
+ IInstanceConfig config = cluster.newInstanceConfig()
+ .set("auto_bootstrap", true)
+ .set(Constants.KEY_DTEST_FULL_STARTUP, true);
+ IInvokableInstance newInstance = cluster.bootstrap(config);
+ newInstance.startup();
+
+ assertEquals("dc1/rack1", cluster.get(1).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc1/rack1", cluster.get(2).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc2/rack2", newInstance.callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ }
+ }
+
public static class BB
{
static void install(ClassLoader cl, int nodeNumber)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 02/06: Retry without time limit calculates wait time correctly
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 856db2afa03c23e40f739c122d09903c1216c1a3
Author: Sam Tunnicliffe <sa...@apache.org>
AuthorDate: Fri Nov 7 17:29:06 2025 +0000
Retry without time limit calculates wait time correctly
Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-21002
---
CHANGES.txt | 1 +
src/java/org/apache/cassandra/tcm/Processor.java | 36 +++----
src/java/org/apache/cassandra/tcm/Retry.java | 2 +-
test/unit/org/apache/cassandra/tcm/RetryTest.java | 111 ++++++++++++++++++++++
4 files changed, 126 insertions(+), 24 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 75983caa80..bf947bc7e2 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.1
+ * Retry without time limit calculates wait time incorrectly (CASSANDRA-21002)
* Don't submit AlterSchemaStatements which produce no effect locally to the CMS (CASSANDRA-21001)
* Avoid iterating all prepared statements when getting PreparedStatementsCacheSize metric (CASSANDRA-21038)
* Reduce performance impact of TableMetadataRef.get and KeyspaceMetadataRef.get (CASSANDRA-20465)
diff --git a/src/java/org/apache/cassandra/tcm/Processor.java b/src/java/org/apache/cassandra/tcm/Processor.java
index e4e99aa08e..f80277b983 100644
--- a/src/java/org/apache/cassandra/tcm/Processor.java
+++ b/src/java/org/apache/cassandra/tcm/Processor.java
@@ -18,17 +18,17 @@
package org.apache.cassandra.tcm;
-import java.util.concurrent.TimeUnit;
-
import com.codahale.metrics.Meter;
-
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.metrics.TCMMetrics;
+import org.apache.cassandra.service.RetryStrategy;
+import org.apache.cassandra.service.TimeoutStrategy;
import org.apache.cassandra.service.WaitStrategy;
import org.apache.cassandra.tcm.log.Entry;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.config.DatabaseDescriptor.getCmsAwaitTimeout;
-import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public interface Processor
{
@@ -51,28 +51,18 @@ public interface Processor
}
/**
- * Since we are using message expiration for communicating timeouts to CMS nodes, we have to be careful not
- * to overflow the long, since messaging is using only 32 bits for deadlines. To achieve that, we are
- * giving `timeoutNanos` every time we retry, but will retry indefinitely.
+ * To be used only when submitting a STARTUP transformation when a node is restarted with a new set of addresses or
+ * running a new release version.
*/
- private static Retry unsafeRetryIndefinitely()
+ static Retry unsafeRetryIndefinitely()
{
- long timeoutNanos = getCmsAwaitTimeout().to(NANOSECONDS);
Meter retryMeter = TCMMetrics.instance.commitRetries;
- return Retry.withNoTimeLimit(retryMeter, new WaitStrategy()
- {
- @Override
- public long computeWaitUntil(int attempts)
- {
- return nanoTime() + timeoutNanos;
- }
-
- @Override
- public long computeWait(int attempts, TimeUnit units)
- {
- return units.convert(timeoutNanos, NANOSECONDS);
- }
- });
+ DurationSpec.IntMillisecondsBound defaultBackoff = DatabaseDescriptor.getDefaultRetryBackoff();
+ DurationSpec.IntMillisecondsBound defaultMaxBackoff = DatabaseDescriptor.getDefaultMaxRetryBackoff();
+ String spec = (defaultBackoff == null ? "100ms" : defaultBackoff.toMilliseconds() + "ms")
+ + "*attempts <=" + (defaultMaxBackoff == null ? "10s" : defaultMaxBackoff.toMilliseconds() + "ms");
+ WaitStrategy wait = RetryStrategy.parse(spec, TimeoutStrategy.LatencySourceFactory.none());
+ return Retry.withNoTimeLimit(retryMeter, wait);
}
Commit.Result commit(Entry.Id entryId, Transformation transform, Epoch lastKnown, Retry retryPolicy);
diff --git a/src/java/org/apache/cassandra/tcm/Retry.java b/src/java/org/apache/cassandra/tcm/Retry.java
index a66f64e113..ca294aaee0 100644
--- a/src/java/org/apache/cassandra/tcm/Retry.java
+++ b/src/java/org/apache/cassandra/tcm/Retry.java
@@ -104,7 +104,7 @@ public class Retry implements WaitStrategy
return -1;
if (deadlineNanos == Long.MAX_VALUE)
- return wait;
+ return units.convert(wait, TimeUnit.NANOSECONDS);
long now = nanoTime();
wait = Math.min(deadlineNanos - now, wait);
diff --git a/test/unit/org/apache/cassandra/tcm/RetryTest.java b/test/unit/org/apache/cassandra/tcm/RetryTest.java
new file mode 100644
index 0000000000..0c2f52c830
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tcm/RetryTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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;
+
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.codahale.metrics.Meter;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.service.WaitStrategy;
+import org.apache.cassandra.tcm.log.Entry;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.transformations.Startup;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static org.apache.cassandra.tcm.membership.MembershipUtils.node;
+import static org.apache.cassandra.tcm.membership.MembershipUtils.nodeAddresses;
+import static org.junit.Assert.assertEquals;
+
+public class RetryTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(RetryTest.class);
+ private Random random;
+
+ @BeforeClass
+ public static void setupClass()
+ {
+ DatabaseDescriptor.daemonInitialization();
+ }
+
+ @Before
+ public void setup()
+ {
+ long seed = System.nanoTime();
+ logger.info("Running test with seed {}", seed);
+ random = new Random(seed);
+ }
+
+ @Test
+ public void testRetryWithNoTimeLimitObservesTimeUnit()
+ {
+ Meter meter = new Meter();
+ final long waitTimeNanos = Math.abs(random.nextLong());
+ WaitStrategy fixed = new WaitStrategy()
+ {
+ @Override
+ public long computeWaitUntil(int attempts) {throw new UnsupportedOperationException();}
+
+ @Override
+ public long computeWait(int attempts, TimeUnit units)
+ {
+ assertEquals(NANOSECONDS, units);
+ return waitTimeNanos;
+ }
+ };
+
+ Retry retry = Retry.withNoTimeLimit(meter, fixed);
+ long waitTimeMillis = retry.computeWait(1, MILLISECONDS);
+ assertEquals(MILLISECONDS.convert(waitTimeNanos, NANOSECONDS), waitTimeMillis);
+ }
+
+
+ @Test
+ public void testProcessorIndefiniteRetryBehaviour()
+ {
+ new Processor()
+ {
+ @Override
+ public ClusterMetadata fetchLogAndWait(Epoch waitFor, Retry retryPolicy) {return null;}
+
+ @Override
+ public Commit.Result commit(Entry.Id entryId, Transformation transform, Epoch lastKnown, Retry retryPolicy)
+ {
+ // Assert the properties of the Retry provided by the private static Processor::unsafeRetryIndefinitely
+ for (int i = 1; i<1000;i++)
+ {
+ // backoff increases in 100ms steps, up to a max of 10000ms
+ long waitTime = retryPolicy.computeWait(i, MILLISECONDS);
+ assertEquals(Math.min((i+1) * 100, 10000), waitTime);
+ }
+ // Retry indefinitely means no explicit deadline is set
+ assertEquals(Long.MAX_VALUE, retryPolicy.deadlineNanos);
+ return null;
+ }
+ }.commit(new Entry.Id(0L), new Startup(node(random), nodeAddresses(random), NodeVersion.CURRENT), Epoch.EMPTY);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/06: AlterSchemaStatement with no effect locally isn't submitted to CMS
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 44f6a3271888300546172563069372fba617ab9c
Author: Sam Tunnicliffe <sa...@apache.org>
AuthorDate: Fri Nov 7 17:28:58 2025 +0000
AlterSchemaStatement with no effect locally isn't submitted to CMS
Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-21001
---
CHANGES.txt | 1 +
.../statements/schema/AlterSchemaStatement.java | 10 +-
.../cql3/AlterSchemaStatementNoOpTest.java | 173 +++++++++++++++++++++
.../cassandra/cql3/PreparedStatementsTest.java | 91 +++++------
4 files changed, 218 insertions(+), 57 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 5a5022e418..75983caa80 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.1
+ * Don't submit AlterSchemaStatements which produce no effect locally to the CMS (CASSANDRA-21001)
* Avoid iterating all prepared statements when getting PreparedStatementsCacheSize metric (CASSANDRA-21038)
* Reduce performance impact of TableMetadataRef.get and KeyspaceMetadataRef.get (CASSANDRA-20465)
* Improve CMS initialization (CASSANDRA-21036)
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java
index f67f5db0e3..e0e82591e8 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java
@@ -183,12 +183,20 @@ abstract public class AlterSchemaStatement implements CQLStatement.SingleKeyspac
// submission to the CMS, but it can't guarantee that the statement can be applied as-is on every node in the
// cluster, as config can be heterogenous falling back to safe defaults may occur on some nodes.
ClusterMetadata metadata = ClusterMetadata.current();
- apply(metadata);
+ Keyspaces proposed = apply(metadata);
+ KeyspacesDiff localDiff = Keyspaces.diff(metadata.schema.getKeyspaces(), proposed);
+ if (localDiff.isEmpty())
+ return new ResultMessage.Void();
+
ClusterMetadata result = commit(metadata);
KeyspacesDiff diff = Keyspaces.diff(metadata.schema.getKeyspaces(), result.schema.getKeyspaces());
clientWarnings(diff).forEach(ClientWarn.instance::warn);
+ // Even though the preliminary local application produced a non-empty diff, there may have been concurrent
+ // schema transformations that had been committed to the log but not yet enacted locally. So there remains a
+ // possibility that the ultimate result is a no-op. i.e. two identical "CREATE IF NOT EXISTS..." racing from
+ // different coordinators.
if (diff.isEmpty())
return new ResultMessage.Void();
diff --git a/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementNoOpTest.java b/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementNoOpTest.java
new file mode 100644
index 0000000000..0c2651aad8
--- /dev/null
+++ b/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementNoOpTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.cql3;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.triggers.TriggersTest;
+
+import static java.lang.String.format;
+import static org.junit.Assert.assertTrue;
+
+public class AlterSchemaStatementNoOpTest extends CQLTester
+{
+ // Tests that when an AlterSchemaStatement is a no-op according to the local schema representation, then the
+ // schema transformation is not submitted to the CMS and serialized into the cluster metadata log. e.g.
+ // * The statement has an IF NOT EXISTS and the target element is present in local schema
+ // * The statement has IF EXISTS and the target element is not present locally
+ @Test
+ public void testKeyspaceNoOps()
+ {
+ String ks = name();
+ assertNoEpochChange(format("ALTER KEYSPACE IF EXISTS %s WITH replication = {'class': 'SimpleStrategy'}", ks));
+ assertMultipleExecutionSingleEpoch(format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy'}", ks));
+ assertMultipleExecutionSingleEpoch(format("DROP KEYSPACE IF EXISTS %s", ks));
+ }
+
+ @Test
+ public void testTableNoOps()
+ {
+ boolean ddm = DatabaseDescriptor.getDynamicDataMaskingEnabled();
+ try
+ {
+ DatabaseDescriptor.setDynamicDataMaskingEnabled(true);
+ String table = KEYSPACE + '.' + name();
+ assertNoEpochChange(format("ALTER TABLE IF EXISTS %s WITH comment = 'test'", table));
+ assertNoEpochChange(format("ALTER TABLE IF EXISTS %s DROP COMPACT STORAGE", table));
+ assertMultipleExecutionSingleEpoch(format("CREATE TABLE IF NOT EXISTS %s (k int PRIMARY KEY, v int)", table));
+ assertNoEpochChange(format("ALTER TABLE %s ADD IF NOT EXISTS v text", table));
+ assertNoEpochChange(format("ALTER TABLE %s ALTER IF EXISTS foo MASKED WITH mask_default()", table));
+ assertNoEpochChange(format("ALTER TABLE %s ALTER IF EXISTS foo DROP MASKED", table));
+ assertNoEpochChange(format("ALTER TABLE %s DROP IF EXISTS foo", table));
+ // RENAME <col> doesn't currently obey the IF EXISTS clause as an IRE is thrown if the column isn't found
+ // assertNoEpochChange(format("ALTER TABLE %s RENAME IF EXISTS foo TO bar", table));
+ assertMultipleExecutionSingleEpoch(format("DROP TABLE IF EXISTS %s", table));
+ }
+ finally
+ {
+ DatabaseDescriptor.setDynamicDataMaskingEnabled(ddm);
+ }
+ }
+
+ @Test
+ public void testUserTypeNoOps()
+ {
+ String type = KEYSPACE + '.' + name();
+ assertNoEpochChange(format("ALTER TYPE IF EXISTS %s ADD f1 int", type));
+ assertMultipleExecutionSingleEpoch(format("CREATE TYPE IF NOT EXISTS %s (f1 int);", type));
+ assertNoEpochChange(format("ALTER TYPE IF EXISTS %s ADD IF NOT EXISTS f1 int", type));
+ assertNoEpochChange(format("ALTER TYPE IF EXISTS %s RENAME IF EXISTS foo to bar", type));
+ assertMultipleExecutionSingleEpoch(format("DROP TYPE IF EXISTS %s", type));
+ }
+
+ @Test
+ public void testIndexNoOps()
+ {
+ String table = createTable("create table %s (k int primary key, v int)");
+ String index = name();
+ assertMultipleExecutionSingleEpoch(format("CREATE INDEX IF NOT EXISTS %s on %s.%s(v)", index, KEYSPACE, table));
+ assertMultipleExecutionSingleEpoch(format("DROP INDEX IF EXISTS %s.%s", KEYSPACE, index));
+ }
+
+ @Test
+ public void testMaterializedViewNoOps()
+ {
+ String table = KEYSPACE + '.' + createTable("CREATE TABLE %s (k int, c int, v text, PRIMARY KEY(k,c))");
+ String view = KEYSPACE + '.' + name();
+ assertNoEpochChange(format("ALTER MATERIALIZED VIEW IF EXISTS %s WITH comment = 'test'", view));
+ assertMultipleExecutionSingleEpoch(format("CREATE MATERIALIZED VIEW IF NOT EXISTS %s " +
+ "AS SELECT * FROM %s " +
+ "WHERE v IS NOT NULL AND k IS NOT NULL AND c IS NOT NULL " +
+ "PRIMARY KEY (v,k,c)",
+ view, table));
+ assertMultipleExecutionSingleEpoch(format("DROP MATERIALIZED VIEW IF EXISTS %s", view));
+ }
+
+ @Test
+ public void testTriggerNoOps()
+ {
+ String table = createTable("create table %s (k int primary key, v int)");
+ String trigger = name();
+ assertMultipleExecutionSingleEpoch(format("CREATE TRIGGER IF NOT EXISTS %s ON %s.%s USING '%s'",
+ trigger, KEYSPACE, table, TriggersTest.TestTrigger.class.getName()));
+ assertMultipleExecutionSingleEpoch(format("DROP TRIGGER IF EXISTS %s ON %s.%s", trigger, KEYSPACE, table));
+ }
+
+ @Test
+ public void testUserFunctionNoOp()
+ {
+ String function = name();
+ assertMultipleExecutionSingleEpoch(format("CREATE FUNCTION IF NOT EXISTS %s.%s(a int, b int) " +
+ "CALLED ON NULL INPUT " +
+ "RETURNS int " +
+ "LANGUAGE java " +
+ "AS 'return Integer.valueOf((a!=null?a.intValue():0) + b.intValue());'",
+ KEYSPACE, function));
+ assertMultipleExecutionSingleEpoch(format("DROP FUNCTION IF EXISTS %s.%s", KEYSPACE, function));
+ }
+
+ @Test
+ public void testUserAggregateNoOp() throws Throwable
+ {
+ String function = createFunction(KEYSPACE,
+ "double, double",
+ "CREATE OR REPLACE FUNCTION %s(state double, val double) " +
+ "RETURNS NULL ON NULL INPUT " +
+ "RETURNS double " +
+ "LANGUAGE java " +
+ "AS 'return state;';");
+ String aggregate = name();
+ assertMultipleExecutionSingleEpoch(format("CREATE AGGREGATE IF NOT EXISTS %s.%s(double) " +
+ "SFUNC %s " +
+ "STYPE double " +
+ "INITCOND 0",
+ KEYSPACE, aggregate, shortFunctionName(function)));
+ assertMultipleExecutionSingleEpoch(format("DROP AGGREGATE IF EXISTS %s.%s", KEYSPACE, aggregate));
+ }
+
+ private void assertMultipleExecutionSingleEpoch(String cql)
+ {
+ // execute the statement once to ensure it is valid and to set
+ // up the test condition (i.e. the thing exists/doesn't exist)
+ Epoch first = ClusterMetadata.current().epoch;
+ QueryProcessor.executeInternal(cql);
+ Epoch second = ClusterMetadata.current().epoch;
+ assertTrue(second.isAfter(first));
+ // execute again to check this is truly a no-op and the epoch isn't incremented
+ QueryProcessor.executeInternal(cql);
+ Epoch third = ClusterMetadata.current().epoch;
+ assertTrue(third.is(second));
+ }
+
+ private void assertNoEpochChange(String cql)
+ {
+ Epoch first = ClusterMetadata.current().epoch;
+ QueryProcessor.executeInternal(cql);
+ Epoch second = ClusterMetadata.current().epoch;
+ assertTrue(second.is(first));
+ }
+
+ private String name()
+ {
+ return "n" + System.nanoTime();
+ }
+}
diff --git a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
index a4430ff78f..850f64a134 100644
--- a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
+++ b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
@@ -47,9 +47,7 @@ import org.apache.cassandra.serializers.Int32Serializer;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.service.accord.AccordService;
-import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
-import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.transport.SimpleClient;
import org.apache.cassandra.transport.messages.ResultMessage;
@@ -84,25 +82,6 @@ public class PreparedStatementsTest extends CQLTester
ClusterMetadataService.instance().log().waitForHighestConsecutive();
}
- private static void runAndAwaitNextEpoch(Runnable runnable)
- {
- try
- {
- Epoch current = ClusterMetadata.current().epoch;
- runnable.run();
- ClusterMetadataService.instance().awaitAtLeast(Epoch.create(current.getEpoch() + 1));
- }
- catch (Throwable e)
- {
- throw new RuntimeException(e);
- }
- }
-
- private static void sessionSchemaUpdate(Session session, String update)
- {
- runAndAwaitNextEpoch(() -> session.execute(update));
- }
-
@Test
public void testUnqualifiedPreparedSelectOrModificationStatementsEmitWarning()
{
@@ -198,13 +177,13 @@ public class PreparedStatementsTest extends CQLTester
public void testInvalidatePreparedStatementsOnDrop()
{
Session session = sessionNet(ProtocolVersion.V5);
- sessionSchemaUpdate(session, dropKsStatement);
- sessionSchemaUpdate(session, createKsStatement);
+ session.execute(dropKsStatement);
+ session.execute(createKsStatement);
String createTableStatement = "CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".qp_cleanup (id int PRIMARY KEY, cid int, val text) WITH transactional_mode='" + test_unsafe + "';";
String dropTableStatement = "DROP TABLE IF EXISTS " + KEYSPACE + ".qp_cleanup;";
- sessionSchemaUpdate(session, createTableStatement);
+ session.execute(createTableStatement);
String insert = "INSERT INTO " + KEYSPACE + ".qp_cleanup (id, cid, val) VALUES (?, ?, ?)";
PreparedStatement prepared = session.prepare(insert);
@@ -212,18 +191,18 @@ public class PreparedStatementsTest extends CQLTester
PreparedStatement preparedTxn = session.prepare(txn(insert));
preparedTxn.setConsistencyLevel(com.datastax.driver.core.ConsistencyLevel.QUORUM);
- sessionSchemaUpdate(session, dropTableStatement);
- sessionSchemaUpdate(session, createTableStatement);
+ session.execute(dropTableStatement);
+ session.execute(createTableStatement);
updateTxnState();
session.execute(prepared.bind(1, 1, "value"));
session.execute(preparedBatch.bind(2, 2, "value2"));
session.execute(preparedTxn.bind(3, 3, "value3"));
- sessionSchemaUpdate(session, dropTableStatement); // since this is an accord table, need to drop the table before the keyspace
- sessionSchemaUpdate(session, dropKsStatement);
- sessionSchemaUpdate(session, createKsStatement);
- sessionSchemaUpdate(session, createTableStatement);
+ session.execute(dropTableStatement); // since this is an accord table, need to drop the table before the keyspace
+ session.execute(dropKsStatement);
+ session.execute(createKsStatement);
+ session.execute(createTableStatement);
updateTxnState();
// The driver will get a response about the prepared statement being invalid, causing it to transparently
@@ -232,8 +211,8 @@ public class PreparedStatementsTest extends CQLTester
session.execute(prepared.bind(1, 1, "value"));
session.execute(preparedBatch.bind(2, 2, "value2"));
session.execute(preparedTxn.bind(3, 3, "value3"));
- sessionSchemaUpdate(session, dropTableStatement); // since this is an accord table, need to drop the table before the keyspace
- sessionSchemaUpdate(session, dropKsStatement);
+ session.execute(dropTableStatement); // since this is an accord table, need to drop the table before the keyspace
+ session.execute(dropKsStatement);
}
@Test
@@ -255,9 +234,9 @@ public class PreparedStatementsTest extends CQLTester
String alterTableStatement = "ALTER TABLE " + KEYSPACE + ".qp_cleanup ADD d int;";
String dropTableStatement = "DROP TABLE IF EXISTS " + KEYSPACE + ".qp_cleanup;";
- sessionSchemaUpdate(session, dropKsStatement);
- sessionSchemaUpdate(session, createKsStatement);
- sessionSchemaUpdate(session, createTableStatement);
+ session.execute(dropKsStatement);
+ session.execute(createKsStatement);
+ session.execute(createTableStatement);
updateTxnState();
String select = "SELECT * FROM " + KEYSPACE + ".qp_cleanup";
@@ -276,7 +255,7 @@ public class PreparedStatementsTest extends CQLTester
assertRowsNet(session.execute(preparedSelectTxn.bind(2)),
row(2, 3, 4));
- sessionSchemaUpdate(session, alterTableStatement);
+ session.execute(alterTableStatement);
updateTxnState();
session.execute("INSERT INTO " + KEYSPACE + ".qp_cleanup (a, b, c, d) VALUES (?, ?, ?, ?);",
@@ -320,8 +299,8 @@ public class PreparedStatementsTest extends CQLTester
}
}
- sessionSchemaUpdate(session, dropTableStatement);
- sessionSchemaUpdate(session, dropKsStatement);
+ session.execute(dropTableStatement);
+ session.execute(dropKsStatement);
}
@Test
@@ -343,9 +322,9 @@ public class PreparedStatementsTest extends CQLTester
String alterTableStatement = "ALTER TABLE " + KEYSPACE + ".qp_cleanup ADD d int;";
String dropTableStatement = "DROP TABLE IF EXISTS " + KEYSPACE + ".qp_cleanup;";
- sessionSchemaUpdate(session, dropKsStatement);
- sessionSchemaUpdate(session, createKsStatement);
- sessionSchemaUpdate(session, createTableStatement);
+ session.execute(dropKsStatement);
+ session.execute(createKsStatement);
+ session.execute(createTableStatement);
updateTxnState();
String select = "SELECT a, b, c FROM " + KEYSPACE + ".qp_cleanup";
@@ -369,7 +348,7 @@ public class PreparedStatementsTest extends CQLTester
Assertions.assertThat(columnNames(rs)).containsExactlyInAnyOrder("a", "b", "c");
}
- sessionSchemaUpdate(session, alterTableStatement);
+ session.execute(alterTableStatement);
updateTxnState();
session.execute("INSERT INTO " + KEYSPACE + ".qp_cleanup (a, b, c, d) VALUES (?, ?, ?, ?);",
@@ -389,8 +368,8 @@ public class PreparedStatementsTest extends CQLTester
Assertions.assertThat(columnNames(rs)).containsExactlyInAnyOrder("a", "b", "c");
}
- sessionSchemaUpdate(session, dropTableStatement);
- sessionSchemaUpdate(session, dropKsStatement);
+ session.execute(dropTableStatement);
+ session.execute(dropKsStatement);
}
@Test
@@ -399,9 +378,9 @@ public class PreparedStatementsTest extends CQLTester
Session session = sessionNet(ProtocolVersion.V5);
session.execute("USE " + keyspace());
- sessionSchemaUpdate(session, dropKsStatement);
- sessionSchemaUpdate(session, createKsStatement);
- runAndAwaitNextEpoch(() -> createTable("CREATE TABLE %s (id int PRIMARY KEY, cid int, val text) WITH transactional_mode='" + test_unsafe + "';"));
+ session.execute(dropKsStatement);
+ session.execute(createKsStatement);
+ createTable("CREATE TABLE %s (id int PRIMARY KEY, cid int, val text) WITH transactional_mode='" + test_unsafe + "';");
updateTxnState();
String insertCQL = "INSERT INTO " + currentTable() + " (id, cid, val) VALUES (?, ?, ?)";
@@ -447,14 +426,14 @@ public class PreparedStatementsTest extends CQLTester
{
Session session = sessionNet(ProtocolVersion.V5);
- sessionSchemaUpdate(session, dropKsStatement);
- sessionSchemaUpdate(session, createKsStatement);
+ session.execute(dropKsStatement);
+ session.execute(createKsStatement);
String table = "custom_expr_test";
String index = "custom_index";
- sessionSchemaUpdate(session, String.format("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY, cid int, val text) WITH transactional_mode='" + test_unsafe + "';",
+ session.execute(String.format("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY, cid int, val text) WITH transactional_mode='" + test_unsafe + "';",
KEYSPACE, table));
- sessionSchemaUpdate(session, String.format("CREATE CUSTOM INDEX %s ON %s.%s(val) USING '%s'",
+ session.execute(String.format("CREATE CUSTOM INDEX %s ON %s.%s(val) USING '%s'",
index, KEYSPACE, table, StubIndex.class.getName()));
updateTxnState();
@@ -493,7 +472,7 @@ public class PreparedStatementsTest extends CQLTester
// Note: this test does not cover all aspects of 10786 (yet) - it was intended to test the
// changes for CASSANDRA-13992.
- runAndAwaitNextEpoch(() -> createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk))"));
+ createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk))");
execute("INSERT INTO %s (pk, v1, v2) VALUES (1,1,1)");
try (SimpleClient simpleClient = newSimpleClient(ProtocolVersion.BETA.orElse(ProtocolVersion.CURRENT)))
@@ -679,7 +658,7 @@ public class PreparedStatementsTest extends CQLTester
{
Session session = sessionNet(version);
session.execute("USE " + keyspace());
- runAndAwaitNextEpoch(() -> createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk))"));
+ createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk))");
PreparedStatement prepared1 = session.prepare(String.format("UPDATE %s SET v1 = ?, v2 = ? WHERE pk = 1 IF v1 = ?", currentTable()));
PreparedStatement prepared2 = session.prepare(String.format("INSERT INTO %s (pk, v1, v2) VALUES (?, 200, 300) IF NOT EXISTS", currentTable()));
@@ -743,7 +722,7 @@ public class PreparedStatementsTest extends CQLTester
{
Session session = sessionNet(version);
session.execute("USE " + keyspace());
- runAndAwaitNextEpoch(() -> createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk))"));
+ createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk))");
PreparedStatement prepared1 = session.prepare("BEGIN BATCH " +
"UPDATE " + currentTable() + " SET v1 = ? WHERE pk = 1 IF v1 = ?;" +
@@ -778,7 +757,7 @@ public class PreparedStatementsTest extends CQLTester
row(false, 1, 10, 20));
assertEquals(rs.getColumnDefinitions().size(), 4);
- runAndAwaitNextEpoch(() -> alterTable("ALTER TABLE %s ADD v3 int;"));
+ alterTable("ALTER TABLE %s ADD v3 int;");
rs = session.execute(prepared2.bind());
assertRowsNet(rs,
@@ -810,7 +789,7 @@ public class PreparedStatementsTest extends CQLTester
int maxAttempts = 3;
Session session = sessionNet(version);
session.execute("USE " + keyspace());
- runAndAwaitNextEpoch(() -> createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk)) WITH transactional_mode='full'"));
+ createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk)) WITH transactional_mode='full'");
updateTxnState();
PreparedStatement writeOnly = session.prepare(txn(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 03/06: Gossip entries for hibernating non-members don't block truncate
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 62bd9ab1ae206a3582eebbff9dbfa8855f479333
Author: Sam Tunnicliffe <sa...@apache.org>
AuthorDate: Fri Nov 7 17:29:21 2025 +0000
Gossip entries for hibernating non-members don't block truncate
Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-21003
---
CHANGES.txt | 1 +
src/java/org/apache/cassandra/gms/Gossiper.java | 5 ++
.../test/HibernatingGossipOnlyMembersTest.java | 67 ++++++++++++++++++++++
3 files changed, 73 insertions(+)
diff --git a/CHANGES.txt b/CHANGES.txt
index bf947bc7e2..0f7cf787f4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.1
+ * Gossip entries for hibernating non-members don't block truncate (CASSANDRA-21003)
* Retry without time limit calculates wait time incorrectly (CASSANDRA-21002)
* Don't submit AlterSchemaStatements which produce no effect locally to the CMS (CASSANDRA-21001)
* Avoid iterating all prepared statements when getting PreparedStatementsCacheSize metric (CASSANDRA-21038)
diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java
index d6aebf1ba8..386bb0e13a 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -402,6 +402,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
for (InetAddressAndPort endpoint : unreachableEndpoints.keySet())
{
NodeId nodeId = metadata.directory.peerId(endpoint);
+ // If the endpoint is not known to cluster metadata, the peer is probably a leftover of upgrading as
+ // long-gone hibernating peers may persist indefinitely. By definition though, such a peer cannot be
+ // a token owner, so it is safe to exclude it.
+ if (nodeId == null)
+ continue;
NodeState state = metadata.directory.peerState(nodeId);
switch (state)
{
diff --git a/test/distributed/org/apache/cassandra/distributed/test/HibernatingGossipOnlyMembersTest.java b/test/distributed/org/apache/cassandra/distributed/test/HibernatingGossipOnlyMembersTest.java
new file mode 100644
index 0000000000..4a0c1dd039
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/HibernatingGossipOnlyMembersTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.gms.ApplicationState;
+import org.apache.cassandra.gms.EndpointState;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.gms.HeartBeatState;
+import org.apache.cassandra.gms.VersionedValue;
+import org.apache.cassandra.locator.InetAddressAndPort;
+
+public class HibernatingGossipOnlyMembersTest extends TestBaseImpl
+{
+ @Test
+ public void testTruncationWithHibernatingGossipOnlyNodes() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+ cluster.get(1).runOnInstance(injectPeerIntoGossip("127.0.0.99"));
+ // Even though the hibernating peer is considered unreachable, it won't prevent the truncation
+ cluster.coordinator(1).execute("TRUNCATE " + KEYSPACE + ".tbl", ConsistencyLevel.ALL);
+ }
+ }
+
+ /**
+ * Add a hibernating peer to gossip which is not a member of the cluster. This can be a legacy
+ * of upgrades, where previously this kind of non-member were never purged from gossip.
+ */
+ private IIsolatedExecutor.SerializableRunnable injectPeerIntoGossip(String address)
+ {
+ return () -> {
+ Map<ApplicationState, VersionedValue> appState = new EnumMap<>(ApplicationState.class);
+ appState.put(ApplicationState.STATUS_WITH_PORT, VersionedValue.unsafeMakeVersionedValue("hibernate,true", 1));
+ EndpointState epState = new EndpointState(HeartBeatState.empty(), appState);
+ Map<InetAddressAndPort, EndpointState> hibernating = new HashMap<>();
+ hibernating.put(InetAddressAndPort.getByNameUnchecked(address), epState);
+ Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.applyStateLocally(hibernating));
+ };
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 06/06: Reject PrepareJoin if tokens already assigned
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 209288cebecba15b320ccd7dd041adab78ce0ff7
Author: Sam Tunnicliffe <sa...@apache.org>
AuthorDate: Mon Nov 10 09:15:42 2025 +0000
Reject PrepareJoin if tokens already assigned
Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-21006
---
CHANGES.txt | 1 +
.../org/apache/cassandra/tcm/ClusterMetadata.java | 5 +-
.../apache/cassandra/tcm/membership/Directory.java | 3 +-
.../cassandra/tcm/transformations/PrepareJoin.java | 13 +++
.../service/accord/AccordTopologyUtils.java | 2 +-
.../reads/repair/AbstractReadRepairTest.java | 6 +-
.../tcm/transformations/PrepareJoinTest.java | 130 +++++++++++++++++++++
.../tcm/transformations/PrepareLeaveTest.java | 26 ++++-
8 files changed, 173 insertions(+), 13 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 7b7ff72e81..f0d05ba26e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.1
+ * Reject PrepareJoin if tokens are already assigned (CASSANDRA-21006)
* Don't update registration status if node state for decommissioned peer is found with the same address (CASSANDRA-21005)
* Avoid NPE when meta keyspace placements are empty before CMS is initialized (CASSANDRA-21004)
* Gossip entries for hibernating non-members don't block truncate (CASSANDRA-21003)
diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
index 65cb869faf..8c962ae299 100644
--- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
+++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
@@ -480,9 +480,10 @@ public class ClusterMetadata
return this;
}
- public Transformer register(NodeId nodeId, NodeAddresses addresses, Location location, NodeVersion version)
+ @VisibleForTesting
+ public Transformer unsafeRegisterForTesting(NodeId nodeId, NodeAddresses addresses, Location location, NodeVersion version)
{
- directory = directory.with(nodeId, addresses, location, version);
+ directory = directory.unsafeWithNodeForTesting(nodeId, addresses, location, version);
return this;
}
diff --git a/src/java/org/apache/cassandra/tcm/membership/Directory.java b/src/java/org/apache/cassandra/tcm/membership/Directory.java
index b7d8684976..8e73f7f341 100644
--- a/src/java/org/apache/cassandra/tcm/membership/Directory.java
+++ b/src/java/org/apache/cassandra/tcm/membership/Directory.java
@@ -178,7 +178,8 @@ public class Directory implements MetadataValue<Directory>
return with(addresses, location, CURRENT);
}
- public Directory with(NodeId id, NodeAddresses addresses, Location location, NodeVersion nodeVersion)
+ @VisibleForTesting
+ public Directory unsafeWithNodeForTesting(NodeId id, NodeAddresses addresses, Location location, NodeVersion nodeVersion)
{
return with(addresses, id, id.toUUID(), location, nodeVersion);
}
diff --git a/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java b/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
index 02fd3d9483..55220e0cea 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.tcm.transformations;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
+import java.util.stream.Collectors;
import com.google.common.collect.ImmutableSet;
@@ -30,6 +31,7 @@ import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Transformation;
@@ -132,6 +134,17 @@ public class PrepareJoin implements Transformation
if (!ALLOWED_STATES.contains(prev.directory.peerState(nodeId)))
return new Rejected(INVALID, String.format("Rejecting this plan as the node %s is in state %s",
nodeId, prev.directory.peerState(nodeId)));
+ Set<Token> alreadyAssigned = prev.tokenMap.tokens().stream().filter(tokens::contains).collect(Collectors.toSet());
+ if (!alreadyAssigned.isEmpty())
+ {
+ String assignedString = alreadyAssigned.stream()
+ .map(t -> {
+ NodeId n = prev.tokenMap.owner(t);
+ InetAddressAndPort e = prev.directory.endpoint(n);
+ return String.format("%s (node %s|%s)", t, n.id(), e);
+ }).collect(Collectors.joining(","));
+ return new Rejected(INVALID, String.format("Rejecting this plan as some tokens are already assigned: [%s]", assignedString));
+ }
PlacementTransitionPlan transitionPlan = placementProvider.planForJoin(prev, nodeId, tokens, prev.schema.getKeyspaces());
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTopologyUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTopologyUtils.java
index e4a2a4791c..2660f07096 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordTopologyUtils.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordTopologyUtils.java
@@ -81,7 +81,7 @@ public class AccordTopologyUtils
NodeId nodeId = nodeId(node);
InetAddressAndPort ep = ep(node);
NodeAddresses addresses = new NodeAddresses(nodeId.toUUID(), ep, ep, ep);
- transformer.register(nodeId, addresses, LOCATION, NodeVersion.CURRENT);
+ transformer.unsafeRegisterForTesting(nodeId, addresses, LOCATION, NodeVersion.CURRENT);
transformer.withNodeState(nodeId, NodeState.JOINED);
transformer.proposeToken(nodeId, Collections.singleton(token));
transformer.addToRackAndDC(nodeId);
diff --git a/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java b/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java
index 4a6d31447c..4e206fe5b5 100644
--- a/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java
@@ -254,9 +254,9 @@ public abstract class AbstractReadRepairTest
ClusterMetadataTestHelper.addEndpoint(replica1.endpoint(), new BytesToken(new byte[] { 0 }), dataCenter1, rack);
ClusterMetadataTestHelper.addEndpoint(replica2.endpoint(), new BytesToken(new byte[] { 1 }), dataCenter1, rack);
ClusterMetadataTestHelper.addEndpoint(replica3.endpoint(), new BytesToken(new byte[] { 2 }), dataCenter1, rack);
- ClusterMetadataTestHelper.addEndpoint(remoteReplica1.endpoint(), new BytesToken(new byte[] { 0 }), dataCenter2, rack);
- ClusterMetadataTestHelper.addEndpoint(remoteReplica2.endpoint(), new BytesToken(new byte[] { 1 }), dataCenter2, rack);
- ClusterMetadataTestHelper.addEndpoint(remoteReplica3.endpoint(), new BytesToken(new byte[] { 2 }), dataCenter2, rack);
+ ClusterMetadataTestHelper.addEndpoint(remoteReplica1.endpoint(), new BytesToken(new byte[] { 3 }), dataCenter2, rack);
+ ClusterMetadataTestHelper.addEndpoint(remoteReplica2.endpoint(), new BytesToken(new byte[] { 4 }), dataCenter2, rack);
+ ClusterMetadataTestHelper.addEndpoint(remoteReplica3.endpoint(), new BytesToken(new byte[] { 5 }), dataCenter2, rack);
for (Replica replica : replicas)
{
diff --git a/test/unit/org/apache/cassandra/tcm/transformations/PrepareJoinTest.java b/test/unit/org/apache/cassandra/tcm/transformations/PrepareJoinTest.java
new file mode 100644
index 0000000000..33ccb137ae
--- /dev/null
+++ b/test/unit/org/apache/cassandra/tcm/transformations/PrepareJoinTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.transformations;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+
+import com.google.common.collect.Sets;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.exceptions.ExceptionCode;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.Transformation;
+import org.apache.cassandra.tcm.membership.Directory;
+import org.apache.cassandra.tcm.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeState;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.ownership.OwnershipUtils;
+
+import static org.apache.cassandra.tcm.membership.MembershipUtils.nodeAddresses;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class PrepareJoinTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(PrepareJoinTest.class);
+ private Random random;
+ private IPartitioner partitioner;
+ private NodeId other;
+ private NodeId joining;
+
+ @Before
+ public void setup()
+ {
+ long seed = System.nanoTime();
+ logger.info("Running test with seed {}", seed);
+ random = new Random(seed);
+ partitioner = Murmur3Partitioner.instance;
+ other = new NodeId(1);
+ joining = new NodeId(2);
+ }
+
+ @Test
+ public void singletonTokenAlreadyAssigned()
+ {
+ ClusterMetadata metadata = metadata();
+ Collection<Token> assigned = metadata.tokenMap.tokens(other);
+ Set<Token> toJoin = Collections.singleton(assigned.iterator().next());
+
+ PrepareJoin prepare = new PrepareJoin(joining, toJoin, PrepareLeaveTest.dummyPlacementProvider, true, true);
+ Transformation.Result result = prepare.execute(metadata);
+ assertTrue(result.isRejected());
+ assertEquals(ExceptionCode.INVALID, result.rejected().code);
+ assertTrue(result.rejected().reason.startsWith("Rejecting this plan as some tokens are already assigned"));
+ }
+
+ @Test
+ public void multipleTokensAlreadyAssigned()
+ {
+ ClusterMetadata metadata = metadata();
+ Collection<Token> assigned = metadata.tokenMap.tokens(other);
+ Set<Token> toJoin = new HashSet<>(assigned);
+
+ PrepareJoin prepare = new PrepareJoin(joining, toJoin, PrepareLeaveTest.dummyPlacementProvider, true, true);
+ Transformation.Result result = prepare.execute(metadata);
+ assertTrue(result.isRejected());
+ assertEquals(ExceptionCode.INVALID, result.rejected().code);
+ assertTrue(result.rejected().reason.startsWith("Rejecting this plan as some tokens are already assigned"));
+ }
+
+ @Test
+ public void noTokensAlreadyAssigned()
+ {
+ ClusterMetadata metadata = metadata();
+ Collection<Token> assigned = metadata.tokenMap.tokens(other);
+ Set<Token> toJoin = OwnershipUtils.randomTokens(16, partitioner, random);
+ while (!Sets.intersection(new HashSet<>(assigned), toJoin).isEmpty())
+ toJoin = OwnershipUtils.randomTokens(16, partitioner, random);
+
+ PrepareJoin prepare = new PrepareJoin(joining, toJoin, PrepareLeaveTest.dummyPlacementProvider, true, true);
+ Transformation.Result result = prepare.execute(metadata);
+ assertTrue(result.isSuccess());
+ }
+
+ private ClusterMetadata metadata()
+ {
+ partitioner = Murmur3Partitioner.instance;
+ other = new NodeId(1);
+ joining = new NodeId(2);
+ Location location = new Location("dc", "rack");
+ Directory directory = new Directory().unsafeWithNodeForTesting(other, nodeAddresses(random), location, NodeVersion.CURRENT)
+ .withNodeState(other, NodeState.JOINED)
+ .unsafeWithNodeForTesting(joining, nodeAddresses(random), location, NodeVersion.CURRENT)
+ .withNodeState(joining, NodeState.REGISTERED);
+ Set<Token> ownedTokens = OwnershipUtils.randomTokens(16, partitioner, random);
+ return ClusterMetadataTestHelper.minimalForTesting(partitioner)
+ .transformer()
+ .with(directory)
+ .proposeToken(other, ownedTokens)
+ .build().metadata;
+ }
+}
diff --git a/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java b/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java
index 4607fa9618..a1b53119fd 100644
--- a/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java
+++ b/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java
@@ -151,27 +151,41 @@ public class PrepareLeaveTest
public static PlacementProvider dummyPlacementProvider = new PlacementProvider()
{
@Override
- public DataPlacements calculatePlacements(Epoch epoch, List<Range<Token>> ranges, ClusterMetadata metadata, Keyspaces keyspaces) { return null; }
+ public DataPlacements calculatePlacements(Epoch epoch, List<Range<Token>> ranges, ClusterMetadata metadata, Keyspaces keyspaces)
+ {
+ return null;
+ }
@Override
- public PlacementTransitionPlan planForJoin(ClusterMetadata metadata, NodeId nodeId, Set<Token> tokens, Keyspaces keyspaces) { return null;}
+ public PlacementTransitionPlan planForJoin(ClusterMetadata metadata, NodeId nodeId, Set<Token> tokens, Keyspaces keyspaces)
+ {
+ return noop();
+ }
@Override
public PlacementTransitionPlan planForMove(ClusterMetadata metadata, NodeId nodeId, Set<Token> tokens, Keyspaces keyspaces)
{
- return null;
+ return noop();
}
@Override
public PlacementTransitionPlan planForDecommission(ClusterMetadata metadata, NodeId nodeId, Keyspaces keyspaces)
+ {
+ return noop();
+ }
+
+ @Override
+ public PlacementTransitionPlan planForReplacement(ClusterMetadata metadata, NodeId replaced, NodeId replacement, Keyspaces keyspaces)
+ {
+ return noop();
+ }
+
+ private PlacementTransitionPlan noop()
{
return new PlacementTransitionPlan(PlacementDeltas.empty(),
PlacementDeltas.empty(),
PlacementDeltas.empty(),
PlacementDeltas.empty());
}
-
- @Override
- public PlacementTransitionPlan planForReplacement(ClusterMetadata metadata, NodeId replaced, NodeId replacement, Keyspaces keyspaces) { return null; }
};
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 04/06: Avoid NPE when meta keyspace placements are empty before CMS is initialized
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit ec7794f20fc0cd07a37901a43f07f4e65517cb08
Author: Sam Tunnicliffe <sa...@apache.org>
AuthorDate: Fri Nov 7 17:29:35 2025 +0000
Avoid NPE when meta keyspace placements are empty before CMS is initialized
Patch by Sam Tunnicliffe; reviewed by Marcus Eriksson for CASSANDRA-21004
---
CHANGES.txt | 1 +
src/java/org/apache/cassandra/service/StorageService.java | 8 +++++---
.../distributed/upgrade/ClusterMetadataSingleNodeUpgradeTest.java | 5 +++++
3 files changed, 11 insertions(+), 3 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 0f7cf787f4..ac61998c3a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.1
+ * Avoid NPE when meta keyspace placements are empty before CMS is initialized (CASSANDRA-21004)
* Gossip entries for hibernating non-members don't block truncate (CASSANDRA-21003)
* Retry without time limit calculates wait time incorrectly (CASSANDRA-21002)
* Don't submit AlterSchemaStatements which produce no effect locally to the CMS (CASSANDRA-21001)
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 84a61f6b26..5f41f57c5c 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -200,6 +200,7 @@ import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.migration.GossipCMSListener;
+import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.MovementMap;
import org.apache.cassandra.tcm.ownership.TokenMap;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
@@ -2070,9 +2071,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
if (keyspaceMetadata.params.replication.isMeta())
{
- rangeToEndpointMap.put(MetaStrategy.entireRange,
- metadata.placements.get(keyspaceMetadata.params.replication)
- .reads.forRange(MetaStrategy.entireRange).get());
+ DataPlacement placement = metadata.placements.get(keyspaceMetadata.params.replication);
+ // May be empty if mid-upgrade and CMS is not yet initialized
+ if (!placement.reads.isEmpty())
+ rangeToEndpointMap.put(MetaStrategy.entireRange, placement.reads.forRange(MetaStrategy.entireRange).get());
}
else
{
diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataSingleNodeUpgradeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataSingleNodeUpgradeTest.java
index 10d6c02dcc..24cf020a73 100644
--- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataSingleNodeUpgradeTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataSingleNodeUpgradeTest.java
@@ -22,6 +22,10 @@ import org.junit.Test;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.service.StorageService;
+
+import static org.junit.Assert.assertTrue;
/**
* @see org.apache.cassandra.tools.nodetool.CMSAdmin.InitializeCMS
@@ -42,6 +46,7 @@ public class ClusterMetadataSingleNodeUpgradeTest extends UpgradeTestBase
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
})
.runAfterClusterUpgrade((cluster) -> {
+ assertTrue(((IInvokableInstance)cluster.get(1)).callOnInstance(() -> StorageService.instance.getRangeToAddressMap("system_cluster_metadata").isEmpty()));
cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
// make sure we can execute transformations:
cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl with comment = 'hello123'"));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org