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 45afe53827 Block decommission if node is a participant in a CMS
reconfiguration
45afe53827 is described below
commit 45afe538274ae70a52a3c915d55e7c8f700ff383
Author: nivy <[email protected]>
AuthorDate: Thu Sep 17 18:12:35 2026 -0400
Block decommission if node is a participant in a CMS reconfiguration
Patch by Nivy Kani; reviewed by Sam Tunnicliffe and Caleb Rackliffe for
CASSANDRA-21591
---
CHANGES.txt | 1 +
.../org/apache/cassandra/tcm/ClusterMetadata.java | 3 +
.../cassandra/tcm/sequences/ReconfigureCMS.java | 6 +-
.../tcm/transformations/PrepareLeave.java | 6 ++
.../ConcurrentCMSReconfigurationAndMoveTest.java | 66 ++++++++++++++++
...CMSReconfigurationAndNodeOperationTestBase.java | 91 ++++++++++++++++++++++
...igurationAndReplacementOfJoiningMemberTest.java | 71 +++++++++++++++++
...igurationAndReplacementOfLeavingMemberTest.java | 83 ++++++++++++++++++++
.../distributed/test/log/TestProcessor.java | 24 +++++-
.../tcm/transformations/PrepareLeaveTest.java | 56 +++++++++++++
10 files changed, 403 insertions(+), 4 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 015680a337..4413bd5fc9 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
6.0-alpha3
+ * Block decommission if node is a participant in a CMS reconfiguration
(CASSANDRA-21591)
* Sharding support for non-vector SAI in-memory indexes (CASSANDRA-18216)
* Index hints serialization rejects a count that the validation accepts
(CASSANDRA-21700)
* Do not expose system_metrics virtual keyspace by default (CASSANDRA-21720)
diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
index 1120269c27..16261f9326 100644
--- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
+++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
@@ -595,6 +595,9 @@ public class ClusterMetadata
MultiStepOperation<?> operation,
Directory
directory)
{
+ if (operation.kind() == MultiStepOperation.Kind.RECONFIGURE_CMS)
+ return false;
+
return operation.affectedPeers(directory).contains(peer);
}
diff --git a/src/java/org/apache/cassandra/tcm/sequences/ReconfigureCMS.java
b/src/java/org/apache/cassandra/tcm/sequences/ReconfigureCMS.java
index 267763404c..1055e1ecd6 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/ReconfigureCMS.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/ReconfigureCMS.java
@@ -153,7 +153,11 @@ public class ReconfigureCMS extends
MultiStepOperation<AdvanceCMSReconfiguration
@Override
public Set<NodeId> affectedPeers(Directory directory)
{
- return Set.of();
+ Set<NodeId> affected = new HashSet<>(next.diff.additions);
+ affected.addAll(next.diff.removals);
+ if (next.activeTransition != null)
+ affected.add(next.activeTransition.nodeId);
+ return affected;
}
@Override
diff --git
a/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
b/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
index eb0e9c4a55..870f446794 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareLeave.java
@@ -32,6 +32,7 @@ import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.NodeId;
@@ -41,6 +42,7 @@ import org.apache.cassandra.tcm.ownership.PlacementProvider;
import org.apache.cassandra.tcm.ownership.PlacementTransitionPlan;
import org.apache.cassandra.tcm.sequences.LeaveStreams;
import org.apache.cassandra.tcm.sequences.LockedRanges;
+import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
import org.apache.cassandra.tcm.sequences.UnbootstrapAndLeave;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
@@ -92,6 +94,10 @@ public class PrepareLeave implements Transformation
if (prev.directory.peerState(leaving) != NodeState.JOINED)
return new Rejected(INVALID, String.format("Rejecting this plan as
the node %s is in state %s", leaving, prev.directory.peerState(leaving)));
+ MultiStepOperation<?> cmsReconfiguration =
prev.inProgressSequences.get(ReconfigureCMS.SequenceKey.instance);
+ if (cmsReconfiguration != null &&
cmsReconfiguration.affectedPeers(prev.directory).contains(leaving))
+ return new Rejected(INVALID, String.format("Rejecting this plan as
the node %s is part of an in-progress CMS reconfiguration: %s", leaving,
cmsReconfiguration));
+
ClusterMetadata proposed =
prev.transformer().proposeRemoveNode(leaving).build().metadata;
if (!force && !validateReplicationForDecommission(proposed))
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndMoveTest.java
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndMoveTest.java
new file mode 100644
index 0000000000..fa41ddf712
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndMoveTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.log;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import com.google.common.collect.Iterables;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.service.StorageService;
+
+public class ConcurrentCMSReconfigurationAndMoveTest extends
ConcurrentCMSReconfigurationAndNodeOperationTestBase
+{
+ @Test
+ public void test() throws Exception
+ {
+ try (Cluster cluster = newCluster())
+ {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IInvokableInstance firstCMS = cluster.get(1);
+ initialCMSSetup(cluster, firstCMS, 2);
+
+ // nodes 1 & 2 are now CMS members, prepare another
reconfiguration which will add node 3
+ // but force it to halt before the first step is committed by
pausing commits on the node
+ // driving it (node1)
+ Future<NodeToolResult> reconfigureResult =
beginAndPauseReconfiguration(executor, firstCMS, 3, "ADDITIONS: [/127.0.0.3");
+
+ // Now the reconfiguration which is active in cluster metadata
includes node3, try moving it to a new token.
+ // node1 is still paused so cannot receive commit requests itself,
but it can still participate in consensus
+ // decisions with node2
+ IInvokableInstance nodeToMove = cluster.get(3);
+ String token = nodeToMove.callsOnInstance(() ->
Iterables.getOnlyElement(StorageService.instance.getLocalTokens()).toString()).call();
+ long moveTo = Long.parseLong(token) + 1;
+ nodeToMove.nodetoolResult("move",
Long.toString(moveTo)).asserts().success();
+ ClusterUtils.waitForCMSToQuiesce(cluster, nodeToMove);
+
+ // now that node3 has completed its token move, unpause node1 to
allow it to continue with the inflight
+ // CMS reconfiguration, which will succeed.
+ ClusterUtils.unpauseCommits(firstCMS);
+ reconfigureResult.get().asserts().success();
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndNodeOperationTestBase.java
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndNodeOperationTestBase.java
new file mode 100644
index 0000000000..29fa5992d8
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndNodeOperationTestBase.java
@@ -0,0 +1,91 @@
+/*
+ * 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.log;
+
+import java.io.IOException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+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.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.tcm.Epoch;
+
+import static
org.apache.cassandra.tcm.Transformation.Kind.ADVANCE_CMS_RECONFIGURATION;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Common scaffolding for tests which pause an in-flight CMS reconfiguration
mid-commit and then
+ * perform some other node operation (replacement, move, ...) concurrently,
before unpausing and
+ * observing the outcome of the reconfiguration.
+ */
+public abstract class ConcurrentCMSReconfigurationAndNodeOperationTestBase
extends TestBaseImpl
+{
+ protected Cluster newCluster() throws IOException
+ {
+ // Configure a short cms_await_timeout as one of the CMS nodes will be
paused, meaning
+ // any commit requests sent to it will eventually timeout and be
resubmitted to another
+ // member. Also minimize the number of acks required by
ProgressBarrier, as the paused
+ // CMS member may not respond to watermark requests if its
INTERNAL_METADATA stage is
+ // filled by backed up commit requests.
+ TokenSupplier even = TokenSupplier.evenlyDistributedTokens(3);
+ return init(builder().withNodes(3)
+ .withTokenSupplier(node -> even.token(node == 4 ?
3 : node))
+ .withoutVNodes()
+ .withConfig(c -> c.with(Feature.GOSSIP,
Feature.NETWORK)
+ .set("cms_await_timeout", "1s")
+
.set("progress_barrier_default_consistency_level", ConsistencyLevel.ONE))
+ .start());
+ }
+
+ /**
+ * Set up the CMS ready to begin the test
+ */
+ protected void initialCMSSetup(Cluster cluster, IInvokableInstance driver,
int newSize)
+ {
+ ClusterUtils.waitForCMSToQuiesce(cluster, driver);
+ driver.nodetoolResult("cms", "reconfigure",
Integer.toString(newSize)).asserts().success();
+ ClusterUtils.waitForCMSToQuiesce(cluster, driver);
+ }
+
+ /**
+ * Prepares another reconfiguration and forces it to halt before its first
step is committed,
+ * by pausing commits on the node driving it, so the caller can perform a
node operation while
+ * the reconfiguration is stalled mid-flight.
+ */
+ protected Future<NodeToolResult>
beginAndPauseReconfiguration(ExecutorService executor,
+
IInvokableInstance driver,
+ int newSize,
+ String
expectedStatus) throws Exception
+ {
+ Callable<Epoch> pending = ClusterUtils.pauseBeforeCommit(driver, e ->
e.kind() == ADVANCE_CMS_RECONFIGURATION);
+ Future<NodeToolResult> inFlightReconfig = executor.submit(() ->
driver.nodetoolResult("cms", "reconfigure", Integer.toString(newSize)));
+ pending.call();
+
+ String status = driver.nodetoolResult("cms", "reconfigure",
"--status").getStdout();
+ assertTrue(status.contains(expectedStatus));
+ return inFlightReconfig;
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndReplacementOfJoiningMemberTest.java
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndReplacementOfJoiningMemberTest.java
new file mode 100644
index 0000000000..7d693943f0
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndReplacementOfJoiningMemberTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.log;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+
+import static org.apache.cassandra.distributed.shared.ClusterUtils.addInstance;
+import static
org.apache.cassandra.distributed.shared.ClusterUtils.awaitRingJoin;
+import static
org.apache.cassandra.distributed.shared.ClusterUtils.startHostReplacement;
+
+public class ConcurrentCMSReconfigurationAndReplacementOfJoiningMemberTest
extends ConcurrentCMSReconfigurationAndNodeOperationTestBase
+{
+ @Test
+ public void test() throws Exception
+ {
+ try (Cluster cluster = newCluster())
+ {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IInvokableInstance firstCMS = cluster.get(1);
+ initialCMSSetup(cluster, firstCMS, 2);
+
+ // nodes 1 & 2 are now CMS members, prepare another
reconfiguration which will add node 3
+ // but force it to halt before the first step is committed by
pausing commits on the node
+ // driving it (node1)
+ Future<NodeToolResult> inFlightReconfig =
beginAndPauseReconfiguration(executor, firstCMS, 3, "ADDITIONS: [/127.0.0.3");
+
+ // Now the reconfiguration which is active in cluster metadata
includes node3, try replacing it.
+ // node1 is still paused so cannot receive commit requests itself,
but it can still participate
+ // in consensus decisions with node2
+ IInvokableInstance nodeToRemove = cluster.get(3);
+ nodeToRemove.shutdown().get();
+ IInvokableInstance replacingNode = addInstance(cluster,
nodeToRemove.config(), c -> c.set("auto_bootstrap", true));
+ startHostReplacement(nodeToRemove, replacingNode, (ignore1_,
ignore2_) -> {});
+ awaitRingJoin(cluster.get(1), replacingNode);
+ awaitRingJoin(replacingNode, cluster.get(1));
+
+ // now that node4 has replaced node3, unpause node1 to allow it to
continue with the inflight CMS
+ // reconfiguration, which will fail
+ ClusterUtils.unpauseCommits(firstCMS);
+ inFlightReconfig.get().asserts().failure();
+ // Cancel the inflight reconfiguration and resubmit another, which
will now succeed
+ firstCMS.nodetoolResult("cms", "reconfigure",
"--cancel").asserts().success();
+ firstCMS.nodetoolResult("cms", "reconfigure",
"3").asserts().success();
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndReplacementOfLeavingMemberTest.java
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndReplacementOfLeavingMemberTest.java
new file mode 100644
index 0000000000..a11b308561
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/log/ConcurrentCMSReconfigurationAndReplacementOfLeavingMemberTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.log;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.Constants;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+
+import static org.apache.cassandra.distributed.shared.ClusterUtils.addInstance;
+import static
org.apache.cassandra.distributed.shared.ClusterUtils.startHostReplacement;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class ConcurrentCMSReconfigurationAndReplacementOfLeavingMemberTest
extends ConcurrentCMSReconfigurationAndNodeOperationTestBase
+{
+ @Test
+ public void test() throws Exception
+ {
+ try (Cluster cluster = newCluster())
+ {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ IInvokableInstance firstCMS = cluster.get(1);
+ initialCMSSetup(cluster, firstCMS, 3);
+
+ // nodes 1, 2 & 3 are now CMS members, prepare another
reconfiguration which will remove node 3
+ // but force it to halt before the first step is committed by
pausing commits on the node
+ // driving it (node1)
+ Future<NodeToolResult> inFlightReconfig =
beginAndPauseReconfiguration(executor, firstCMS, 2, "REMOVALS: [/127.0.0.3");
+
+ // Now the reconfiguration which is active in cluster metadata
includes node3, try replacing it.
+ // node1 is still paused so cannot receive commit requests itself,
but it can still participate
+ // in consensus decisions with node2
+ IInvokableInstance nodeToRemove = cluster.get(3);
+ nodeToRemove.shutdown().get();
+
+ // The startup of the replacement node fails *after*
messaging/gossip threads have started, so we configure it
+ // *not* to mark itself as shutdown on failure because this causes
it to be skipped by the cluster shutdown
+ // at the end of the test. If not properly terminated, the
instance remains technically alive and able to
+ // communicate with instances started by subsequent tests and
interfere with the startup of those other clusters.
+ IInvokableInstance replacingNode = addInstance(cluster,
nodeToRemove.config(),
+ c ->
c.set("auto_bootstrap", true)
+
.set(Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN, false));
+ try
+ {
+ startHostReplacement(nodeToRemove, replacingNode, (ignore1_,
ignore2_) -> {});
+ fail("Instance replacement should fail");
+ }
+ catch (Exception e)
+ {
+ assertTrue(e.getMessage().contains("Can not commit
transformation: \"INVALID\"(Can not add a new in-progress sequence for
Reconfigure CMS"));
+ }
+
+ // the replacement has failed, unpause node1 to allow it to
continue with the inflight CMS
+ // reconfiguration, which will succeed.
+ ClusterUtils.unpauseCommits(firstCMS);
+ inFlightReconfig.get().asserts().success();
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java
b/test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java
index f7f5c50262..8118fe8ae4 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java
@@ -28,6 +28,7 @@ import java.util.function.Predicate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.exceptions.ExceptionCode;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Commit;
import org.apache.cassandra.tcm.Epoch;
@@ -57,8 +58,22 @@ public class TestProcessor implements Processor
@Override
public Commit.Result commit(Entry.Id entryId, Transformation transform,
Epoch lastKnown, Retry retryPolicy)
{
- maybePause(transform);
- waitIfPaused();
+ if (maybePause(transform))
+ {
+ // Block the calling thread, this happens at most once per
registered predicate but if there are many
+ // of these the InternalMetadataStage can become saturated by
paused threads. This can impact other
+ // work, most notably responding to TCM_CURRENT_EPOCH_REQ messages
issued by ProgressBarrier.
+ waitIfPaused();
+ }
+ else if (isPaused())
+ {
+ // Any commit that arrives while already paused must not park the
calling thread, so reject immediately
+ // instead, with a retriable (non-rejected) failure. Callers
should already know how to retry such a
+ // failure.
+ logger.debug("Test processor is paused; rejecting unrelated commit
of {} instead of blocking", transform.kind());
+ return Commit.Result.failed(ExceptionCode.SERVER_ERROR,
+ "TestProcessor is paused for an
unrelated transformation; retry elsewhere");
+ }
Commit.Result commited = delegate.commit(entryId, transform,
lastKnown, retryPolicy);
testCommitPredicates(transform, commited);
return commited;
@@ -106,8 +121,9 @@ public class TestProcessor implements Processor
commitPredicates.add(fn);
}
- private void maybePause(Transformation transform)
+ private boolean maybePause(Transformation transform)
{
+ boolean paused = false;
Iterator<Predicate<Transformation>> iter = waitPredicates.iterator();
while (iter.hasNext())
@@ -116,8 +132,10 @@ public class TestProcessor implements Processor
{
pause();
iter.remove();
+ paused = true;
}
}
+ return paused;
}
private void testCommitPredicates(Transformation transform, Commit.Result
result)
diff --git
a/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java
b/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java
index bbf58eda54..aff2f320c2 100644
--- a/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java
+++ b/test/unit/org/apache/cassandra/tcm/transformations/PrepareLeaveTest.java
@@ -55,6 +55,10 @@ import org.apache.cassandra.tcm.ownership.PlacementDeltas;
import org.apache.cassandra.tcm.ownership.PlacementProvider;
import org.apache.cassandra.tcm.ownership.PlacementTransitionPlan;
import org.apache.cassandra.tcm.sequences.LeaveStreams;
+import org.apache.cassandra.tcm.sequences.LockedRanges;
+import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
+import org.apache.cassandra.tcm.transformations.cms.AdvanceCMSReconfiguration;
+import org.apache.cassandra.tcm.transformations.cms.PrepareCMSReconfiguration;
import static
org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper.addr;
import static org.junit.Assert.assertFalse;
@@ -106,6 +110,58 @@ public class PrepareLeaveTest
assertFalse(executeLeave(metadata));
}
+ @Test
+ public void testRejectedWhenLeavingNodeIsCMSReconfigurationAdd() throws
Throwable
+ {
+ Keyspaces kss =
Keyspaces.of(DistributedMetadataLogKeyspace.initialMetadata(Sets.newHashSet(hostDc.values())),
KSM);
+ ClusterMetadata metadata = prepMetadata(kss, 2, 2);
+ NodeId leaving =
metadata.directory.peerId(InetAddressAndPort.getByName("127.0.0.11"));
+ PrepareCMSReconfiguration.Diff diff = new
PrepareCMSReconfiguration.Diff(Collections.singletonList(leaving),
Collections.emptyList());
+ assertFalse(executeLeave(withReconfiguringCMS(metadata, diff)));
+ }
+
+ @Test
+ public void testRejectedWhenLeavingNodeIsCMSReconfigurationRemove() throws
Throwable
+ {
+ Keyspaces kss =
Keyspaces.of(DistributedMetadataLogKeyspace.initialMetadata(Sets.newHashSet(hostDc.values())),
KSM);
+ ClusterMetadata metadata = prepMetadata(kss, 2, 2);
+ NodeId leaving =
metadata.directory.peerId(InetAddressAndPort.getByName("127.0.0.11"));
+ PrepareCMSReconfiguration.Diff diff = new
PrepareCMSReconfiguration.Diff(Collections.emptyList(),
Collections.singletonList(leaving));
+ assertFalse(executeLeave(withReconfiguringCMS(metadata, diff)));
+ }
+
+ private ClusterMetadata withReconfiguringCMS(ClusterMetadata metadata,
PrepareCMSReconfiguration.Diff diff)
+ {
+ ReconfigureCMS sequence =
ReconfigureCMS.newSequence(LockedRanges.keyFor(Epoch.FIRST), diff);
+ return metadata.transformer()
+
.with(metadata.inProgressSequences.with(ReconfigureCMS.SequenceKey.instance,
sequence))
+ .build()
+ .metadata;
+ }
+
+ @Test
+ public void testRejectedWhenLeavingNodeIsActiveCMSTransition() throws
Throwable
+ {
+ Keyspaces kss =
Keyspaces.of(DistributedMetadataLogKeyspace.initialMetadata(Sets.newHashSet(hostDc.values())),
KSM);
+ ClusterMetadata metadata = prepMetadata(kss, 2, 2);
+ NodeId leaving =
metadata.directory.peerId(InetAddressAndPort.getByName("127.0.0.11"));
+
+ LockedRanges.Key lockKey = LockedRanges.keyFor(Epoch.FIRST);
+ ReconfigureCMS sequence = ReconfigureCMS.newSequence(lockKey,
PrepareCMSReconfiguration.Diff.NOCHANGE);
+ AdvanceCMSReconfiguration activeStep = new AdvanceCMSReconfiguration(0,
+
Epoch.EMPTY,
+
lockKey,
+
PrepareCMSReconfiguration.Diff.NOCHANGE,
+
ReconfigureCMS.ActiveTransition.withNodeIds(leaving, Collections.emptySet()));
+ sequence = sequence.advance(activeStep);
+
+ metadata = metadata.transformer()
+
.with(metadata.inProgressSequences.with(ReconfigureCMS.SequenceKey.instance,
sequence))
+ .build()
+ .metadata;
+ assertFalse(executeLeave(metadata));
+ }
+
private boolean executeLeave(ClusterMetadata metadata) throws Throwable
{
PrepareLeave prepareLeave = new
PrepareLeave(metadata.directory.peerId(InetAddressAndPort.getByName("127.0.0.11")),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
