From af5eed2a3277b747a1a4a2b7bb18dce39fb167d5 Mon Sep 17 00:00:00 2001 From: Marcus Eriksson Date: Mon, 16 Mar 2026 15:04:41 +0100 Subject: [PATCH 1/6] Handle lost response when committing PrepareMove Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-21222 --- CHANGES.txt | 1 + .../org/apache/cassandra/tcm/Startup.java | 2 + .../tcm/sequences/SingleNodeSequences.java | 40 +++++- .../tcm/sequences/UnbootstrapAndLeave.java | 15 +- .../test/tcm/LostCommitReqResTest.java | 135 ++++++++++++++++++ 5 files changed, 180 insertions(+), 13 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/tcm/LostCommitReqResTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 86f663dfdf..fcb0f59a8d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Handle lost response when committing PrepareMove (CASSANDRA-21222) * SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity (CASSANDRA-21429) * Safely regain ranges and delete retired command stores (CASSANDRA-21212) * Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360) diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java index 970ea190c6..2bfe4b4cc9 100644 --- a/src/java/org/apache/cassandra/tcm/Startup.java +++ b/src/java/org/apache/cassandra/tcm/Startup.java @@ -449,6 +449,8 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; if (isReplacing) ReconfigureCMS.maybeReconfigureCMS(metadata, DatabaseDescriptor.getReplaceAddress()); + // if this throws startup is aborted and operator needs to restart, in that case the IPS is resumed if + // it was successfully committed ClusterMetadataService.instance().commit(initialTransformation.get()); // When Accord starts up it needs to check for any historic epochs that it needs to know about (in order // to handle pending transactions), in order to know what nodes to check with it needs to know what the diff --git a/src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java b/src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java index 3a2fd2221a..313700aaaf 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java +++ b/src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java @@ -85,7 +85,9 @@ public interface SingleNodeSequences ClusterMetadataService.instance().commit(new PrepareLeave(self, force, ClusterMetadataService.instance().placementProvider(), - LeaveStreams.Kind.UNBOOTSTRAP)); + LeaveStreams.Kind.UNBOOTSTRAP), + m -> m, + failureHandler("PrepareLeave", StorageService.instance::markDecommissionFailed)); } else if (InProgressSequences.isLeave(inProgress)) { @@ -182,13 +184,24 @@ public interface SingleNodeSequences ClusterMetadataService.instance().commit(new PrepareMove(self, Collections.singleton(newToken), ClusterMetadataService.instance().placementProvider(), - true)); + true), + m -> m, + failureHandler("PrepareMove", StorageService.instance::markMoveFailed)); InProgressSequences.finishInProgressSequences(self); if (logger.isDebugEnabled()) logger.debug("Successfully moved to new token {}", StorageService.instance.getLocalTokens().iterator().next()); } + private static ClusterMetadataService.CommitFailureHandler failureHandler(String type, Runnable markFailed) + { + return (code, msg) -> { + logger.warn("Got failure committing {} transformation: {} {}", type, code, msg); + markFailed.run(); + throw new IllegalStateException(String.format("Can not commit transformation: \"%s\"(%s).", code, msg)); + }; + } + static void resumeMove() { if (ClusterMetadataService.instance().isMigrating() || ClusterMetadataService.state() == ClusterMetadataService.State.GOSSIP) @@ -201,6 +214,11 @@ public interface SingleNodeSequences { String msg = "No move operation in progress, can't resume"; logger.info(msg); + if (StorageService.instance.operationMode() == MOVE_FAILED) + { + // there is no ongoing move to resume, but operation mode thinks there is + StorageService.instance.clearTransientMode(); + } throw new IllegalStateException(msg); } if (StorageService.instance.operationMode() != MOVE_FAILED) @@ -221,7 +239,7 @@ public interface SingleNodeSequences /** * * @param nodeId node id to abort the MSO for, null for local node - * @param kind the expected kind of the multi step operation to abolt + * @param kind the expected kind of the multi step operation to abort * @param ssMode the legacy mode we want storage service to be in, null for any */ private static void abortHelper(@Nullable String nodeId, MultiStepOperation.Kind kind, @Nullable StorageService.Mode ssMode) @@ -234,9 +252,19 @@ public interface SingleNodeSequences MultiStepOperation sequence = metadata.inProgressSequences.get(toAbort); if (sequence == null || sequence.kind() != kind) { - String msg = String.format("No %s operation in progress for %s, can't abort (%s)", kind, toAbort, sequence); - logger.info(msg); - throw new IllegalStateException(msg); + if (toAbort.equals(metadata.myNodeId()) && ssMode != null && StorageService.instance.operationMode() == ssMode) + { + // there is no ongoing sequence with the given kind, but storage service operation mode is set, clear it + logger.debug("There is no ongoing {} sequence for this node, but operation mode is {} - clearing transient mode", kind, ssMode); + StorageService.instance.clearTransientMode(); + return; + } + else + { + String msg = String.format("No %s operation in progress for %s, can't abort (%s)", kind, toAbort, sequence); + logger.info(msg); + throw new IllegalStateException(msg); + } } if (toAbort.equals(metadata.myNodeId())) { diff --git a/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java b/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java index da8ae74e6a..deca30ab92 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java +++ b/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java @@ -92,12 +92,12 @@ public class UnbootstrapAndLeave extends MultiStepOperation */ @VisibleForTesting UnbootstrapAndLeave(Epoch latestModification, - LockedRanges.Key lockKey, - Transformation.Kind next, - PrepareLeave.StartLeave startLeave, - PrepareLeave.MidLeave midLeave, - PrepareLeave.FinishLeave finishLeave, - LeaveStreams streams) + LockedRanges.Key lockKey, + Transformation.Kind next, + PrepareLeave.StartLeave startLeave, + PrepareLeave.MidLeave midLeave, + PrepareLeave.FinishLeave finishLeave, + LeaveStreams streams) { super(nextToIndex(next), latestModification); this.lockKey = lockKey; @@ -198,7 +198,8 @@ public class UnbootstrapAndLeave extends MultiStepOperation } catch (ExecutionException e) { - StorageService.instance.markDecommissionFailed(); + if (startLeave.nodeId().equals(ClusterMetadata.current().myNodeId())) + StorageService.instance.markDecommissionFailed(); JVMStabilityInspector.inspectThrowable(e); logger.error("Error while decommissioning node: {}", e.getCause().getMessage()); throw new RuntimeException("Error while decommissioning node: " + e.getCause().getMessage()); diff --git a/test/distributed/org/apache/cassandra/distributed/test/tcm/LostCommitReqResTest.java b/test/distributed/org/apache/cassandra/distributed/test/tcm/LostCommitReqResTest.java new file mode 100644 index 0000000000..dfa3512d09 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/tcm/LostCommitReqResTest.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test.tcm; + +import java.io.IOException; + +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.net.Verb; +import org.apache.cassandra.service.StorageService; + +import static org.junit.Assert.assertEquals; + +public class LostCommitReqResTest extends TestBaseImpl +{ + @Test + public void lostMoveCommitResponseTest() throws IOException + { + try (Cluster cluster = init(builder().withNodes(2) + .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5")) + .start())) + { + // no commit responses + cluster.filters().verbs(Verb.TCM_COMMIT_RSP.id).from(1).to(2).drop(); + // lost response when committing PrepareMove, fails the nodetool command and halts progress + cluster.get(2).nodetoolResult("move", "1234").asserts().failure(); + assertMoveFailed(cluster.get(2)); // we should be in MOVE_FAILED state to allow abortmove + // still no responses, committing CancelInProgressSequence response is lost, but is actually committed + cluster.get(2).nodetoolResult("abortmove").asserts().failure(); + assertNormal(cluster.get(2)); // and we should be back to normal + cluster.get(2).nodetoolResult("move", "1234").asserts().failure(); + assertMoveFailed(cluster.get(2)); + // finishing the MSO does not depend on any commit responses, just that ClusterMetadata.current() is up to date, so this is successful; + cluster.get(2).nodetoolResult("move", "--resume").asserts().success(); + assertNormal(cluster.get(2)); + } + } + + @Test + public void lostMoveCommitRequestTest() throws IOException + { + try (Cluster cluster = init(builder().withNodes(2) + .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5")) + .start())) + { + // no commit requests + cluster.filters().verbs(Verb.TCM_COMMIT_REQ.id).from(2).to(1).drop(); + cluster.get(2).nodetoolResult("move", "1234").asserts().failure(); + // state should be "move failed" since we don't know if the request or response went missing: + assertMoveFailed(cluster.get(2)); + cluster.filters().reset(); + // abort move should be successful, it only clears the transient state in this case though + cluster.get(2).nodetoolResult("abortmove").asserts().success(); + assertNormal(cluster.get(2)); + cluster.get(2).nodetoolResult("move", "1234").asserts().success(); + assertNormal(cluster.get(2)); + } + } + + @Test + public void lostDecomCommitResponseTest() throws IOException + { + try (Cluster cluster = init(builder().withNodes(2) + .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5")) + .start())) + { + cluster.filters().verbs(Verb.TCM_COMMIT_RSP.id).from(1).to(2).drop(); + cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure(); + assertDecomFailed(cluster.get(2)); + cluster.get(2).nodetoolResult("abortdecommission").asserts().failure(); + assertNormal(cluster.get(2)); + cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure(); + assertDecomFailed(cluster.get(2)); + cluster.get(2).nodetoolResult("decommission").asserts().success(); + } + } + + @Test + public void lostDecomCommitRequestTest() throws IOException + { + try (Cluster cluster = init(builder().withNodes(2) + .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5")) + .start())) + { + cluster.filters().verbs(Verb.TCM_COMMIT_REQ.id).from(2).to(1).drop(); + cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure(); + assertDecomFailed(cluster.get(2)); + cluster.filters().reset(); + cluster.get(2).nodetoolResult("abortdecommission").asserts().success(); + assertNormal(cluster.get(2)); + cluster.get(2).nodetoolResult("decommission", "--force").asserts().success(); + } + } + + private static void assertNormal(IInvokableInstance i) + { + assertOperationMode(i, StorageService.Mode.NORMAL); + } + + private static void assertMoveFailed(IInvokableInstance i) + { + assertOperationMode(i, StorageService.Mode.MOVE_FAILED); + } + + private static void assertDecomFailed(IInvokableInstance i) + { + assertOperationMode(i, StorageService.Mode.DECOMMISSION_FAILED); + } + + private static void assertOperationMode(IInvokableInstance i, StorageService.Mode expectedMode) + { + String mode = i.callOnInstance(() -> StorageService.instance.operationMode().toString()); + assertEquals(expectedMode.toString(), mode); + } +} From b6c274933ea872acb96c229613f7fae4cf92dfae Mon Sep 17 00:00:00 2001 From: Marcus Eriksson Date: Mon, 16 Mar 2026 15:04:41 +0100 Subject: [PATCH 2/6] Minor TokenMap performance improvement Patch by marcuse; reviewed by Dmitry Konstantinov and Sam Tunnicliffe for CASSANDRA-21223 --- CHANGES.txt | 1 + .../cassandra/tcm/ownership/TokenMap.java | 18 +++--------------- 2 files changed, 4 insertions(+), 15 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index fcb0f59a8d..86f388f22d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Minor TokenMap performance improvement (CASSANDRA-21223) * Handle lost response when committing PrepareMove (CASSANDRA-21222) * SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity (CASSANDRA-21429) * Safely regain ranges and delete retired command stores (CASSANDRA-21212) diff --git a/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java b/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java index ed82642c8a..993fded7df 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java +++ b/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java @@ -55,7 +55,7 @@ public class TokenMap implements MetadataValue private static final Logger logger = LoggerFactory.getLogger(TokenMap.class); private final SortedBiMultiValMap map; - private final List tokens; + private final ImmutableList tokens; private final List> ranges; // TODO: move partitioner to the users (SimpleStrategy and Uniform Range Placement?) private final IPartitioner partitioner; @@ -71,7 +71,7 @@ public class TokenMap implements MetadataValue this.lastModified = lastModified; this.partitioner = partitioner; this.map = map; - this.tokens = tokens(); + this.tokens = ImmutableList.copyOf(map.keySet()); this.ranges = toRanges(tokens, partitioner); } @@ -101,18 +101,6 @@ public class TokenMap implements MetadataValue return new TokenMap(lastModified, partitioner, finalisedCopy); } - public TokenMap unassignTokens(NodeId id, Collection tokens) - { - SortedBiMultiValMap finalisedCopy = SortedBiMultiValMap.create(map); - for (Token token : tokens) - { - NodeId nodeId = finalisedCopy.remove(token); - assert nodeId.equals(id); - } - - return new TokenMap(lastModified, partitioner, finalisedCopy); - } - public SortedBiMultiValMap asMap() { return SortedBiMultiValMap.create(map); @@ -130,7 +118,7 @@ public class TokenMap implements MetadataValue public ImmutableList tokens() { - return ImmutableList.copyOf(map.keySet()); + return tokens; } public ImmutableList tokens(NodeId nodeId) From 9af2b2cdf8d8c4a3088673b9d67b0e97518c48e8 Mon Sep 17 00:00:00 2001 From: Marcus Eriksson Date: Mon, 16 Mar 2026 15:04:41 +0100 Subject: [PATCH 3/6] Improve performance deserializing cluster metadata Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-21224 --- CHANGES.txt | 1 + .../cassandra/tcm/membership/Directory.java | 167 +++++++++++------- .../DirectorySerializationBench.java | 126 +++++++++++++ 3 files changed, 234 insertions(+), 60 deletions(-) create mode 100644 test/microbench/org/apache/cassandra/test/microbench/DirectorySerializationBench.java diff --git a/CHANGES.txt b/CHANGES.txt index 86f388f22d..ff76806971 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Improve performance when deserializing cluster metadata (CASSANDRA-21224) * Minor TokenMap performance improvement (CASSANDRA-21223) * Handle lost response when committing PrepareMove (CASSANDRA-21222) * SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity (CASSANDRA-21429) diff --git a/src/java/org/apache/cassandra/tcm/membership/Directory.java b/src/java/org/apache/cassandra/tcm/membership/Directory.java index 8c7942a1fd..55a5e7ac96 100644 --- a/src/java/org/apache/cassandra/tcm/membership/Directory.java +++ b/src/java/org/apache/cassandra/tcm/membership/Directory.java @@ -50,7 +50,6 @@ import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.MetadataValue; import org.apache.cassandra.tcm.serialization.MetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; -import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.UUIDSerializer; import org.apache.cassandra.utils.btree.BTreeBiMap; import org.apache.cassandra.utils.btree.BTreeMap; @@ -58,6 +57,7 @@ import org.apache.cassandra.utils.btree.BTreeMultimap; import static org.apache.cassandra.db.TypeSizes.sizeof; import static org.apache.cassandra.tcm.membership.NodeVersion.CURRENT; +import static org.apache.cassandra.tcm.membership.NodeVersion.CURRENT_METADATA_VERSION; public class Directory implements MetadataValue { @@ -106,6 +106,41 @@ public class Directory implements MetadataValue BTreeMap addresses, BTreeMultimap endpointsByDC, BTreeMap> racksByDC) + { + this(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC, clusterVersions(states, versions)); + } + + private Directory(int nextId, + Epoch lastModified, + BTreeBiMap peers, + BTreeSet removedNodes, + BTreeMap locations, + BTreeMap states, + BTreeMap versions, + BTreeBiMap hostIds, + BTreeMap addresses, + BTreeMultimap endpointsByDC, + BTreeMap> racksByDC, + ClusterVersions clusterVersions) + { + this(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC, + clusterVersions.clusterMinVersion, clusterVersions.clusterMaxVersion, clusterVersions.commonSerializationVersion); + } + + private Directory(int nextId, + Epoch lastModified, + BTreeBiMap peers, + BTreeSet removedNodes, + BTreeMap locations, + BTreeMap states, + BTreeMap versions, + BTreeBiMap hostIds, + BTreeMap addresses, + BTreeMultimap endpointsByDC, + BTreeMap> racksByDC, + NodeVersion clusterMinVersion, + NodeVersion clusterMaxVersion, + Version commonSerializationVersion) { this.nextId = nextId; this.lastModified = lastModified; @@ -118,10 +153,9 @@ public class Directory implements MetadataValue this.addresses = addresses; this.endpointsByDC = endpointsByDC; this.racksByDC = racksByDC; - Pair minMaxVer = minMaxVersions(states, versions); - clusterMinVersion = minMaxVer.left; - clusterMaxVersion = minMaxVer.right; - commonSerializationVersion = minCommonSerializationVersion(states, versions); + this.clusterMinVersion = clusterMinVersion; + this.clusterMaxVersion = clusterMaxVersion; + this.commonSerializationVersion = commonSerializationVersion; } @Override @@ -161,7 +195,7 @@ public class Directory implements MetadataValue @Override public Directory withLastModified(Epoch epoch) { - return new Directory(nextId, epoch, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC); + return new Directory(nextId, epoch, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC, clusterMinVersion, clusterMaxVersion, commonSerializationVersion); } public Directory withNonUpgradedNode(NodeAddresses addresses, @@ -250,9 +284,18 @@ public class Directory implements MetadataValue BTreeMap> updatedEndpointsByRack = racksByDC.withForce(location(id).datacenter, rackEP); return new Directory(nextId, lastModified, - peers.withForce(id,nodeAddresses.broadcastAddress), removedNodes, locations, states, versions, hostIds, addresses.withForce(id, nodeAddresses), + peers.withForce(id, nodeAddresses.broadcastAddress), + removedNodes, + locations, + states, + versions, + hostIds, + addresses.withForce(id, nodeAddresses), updatedEndpointsByDC, - updatedEndpointsByRack); + updatedEndpointsByRack, + clusterMinVersion, + clusterMaxVersion, + commonSerializationVersion); } public Directory withRackAndDC(NodeId id) @@ -266,7 +309,10 @@ public class Directory implements MetadataValue return new Directory(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC.with(location.datacenter, endpoint), - racksByDC.withForce(location.datacenter, rackEP)); + racksByDC.withForce(location.datacenter, rackEP), + clusterMinVersion, + clusterMaxVersion, + commonSerializationVersion); } public Directory withoutRackAndDC(NodeId id) @@ -286,7 +332,10 @@ public class Directory implements MetadataValue newRacksByDC = racksByDC.withForce(location.datacenter, rackEP); return new Directory(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC.without(location.datacenter, endpoint), - newRacksByDC); + newRacksByDC, + clusterMinVersion, + clusterMaxVersion, + commonSerializationVersion); } public Directory withUpdatedRackAndDc(NodeId id, Location location) @@ -306,23 +355,7 @@ public class Directory implements MetadataValue return this; return new Directory(nextId, lastModified, peers, removedNodes, locations.withForce(id, location), states, versions, hostIds, - addresses, endpointsByDC, racksByDC); - } - - public Directory removed(Epoch removedIn, NodeId id, InetAddressAndPort addr) - { - Invariants.require(!peers.containsKey(id)); - return new Directory(nextId, - lastModified, - peers, - removedNodes.with(new RemovedNode(removedIn, id, addr)), - locations, - states, - versions, - hostIds, - addresses, - endpointsByDC, - racksByDC); + addresses, endpointsByDC, racksByDC, clusterMinVersion, clusterMaxVersion, commonSerializationVersion); } public Directory without(Epoch removedIn, NodeId id) @@ -641,14 +674,23 @@ public class Directory implements MetadataValue if (version.isAtLeast(Version.V1)) nextId = in.readInt(); int count = in.readInt(); - Directory newDir = new Directory(); + BTreeBiMap peers = BTreeBiMap.empty(); + BTreeMap locations = BTreeMap.empty(); + BTreeMap states = BTreeMap.empty(); + BTreeMap versions = BTreeMap.empty(); + BTreeBiMap hostIds = BTreeBiMap.empty(); + BTreeMap addresses = BTreeMap.empty(); for (int i = 0; i < count; i++) { Node n = Node.serializer.deserialize(in, version); - // todo: bulk operations - newDir = newDir.with(n.addresses, n.id, n.hostId, n.location, n.version) - .withNodeState(n.id, n.state); + NodeId id = n.id; + peers = peers.withForce(id, n.addresses.broadcastAddress); + locations = locations.withForce(id, n.location); + states = states.withForce(id, n.state); + versions = versions.withForce(id, n.version); + hostIds = hostIds.withForce(id, n.hostId); + addresses = addresses.withForce(id, n.addresses); } int dcCount = in.readInt(); @@ -677,7 +719,7 @@ public class Directory implements MetadataValue if (version.isBefore(Version.V1)) { NodeId maxId = null; - for (NodeId id : newDir.peers.keySet()) + for (NodeId id : peers.keySet()) { if (maxId == null || id.compareTo(maxId) > 0) maxId = id; @@ -688,7 +730,7 @@ public class Directory implements MetadataValue else nextId = maxId.id() + 1; } - + BTreeSet removed = BTreeSet.empty(RemovedNode::compareTo); if (version.isAtLeast(Version.V7)) { int removedNodes = in.readInt(); @@ -697,19 +739,20 @@ public class Directory implements MetadataValue long epoch = in.readLong(); NodeId nodeId = NodeId.serializer.deserialize(in, version); InetAddressAndPort addr = InetAddressAndPort.MetadataSerializer.serializer.deserialize(in, version); - newDir.removed(Epoch.create(epoch), nodeId, addr); + Invariants.require(!peers.containsKey(nodeId)); + removed = removed.with(new RemovedNode(Epoch.create(epoch), nodeId, addr)); } } return new Directory(nextId, lastModified, - newDir.peers, - newDir.removedNodes, - newDir.locations, - newDir.states, - newDir.versions, - newDir.hostIds, - newDir.addresses, + peers, + removed, + locations, + states, + versions, + hostIds, + addresses, dcEndpoints, racksByDC); } @@ -769,10 +812,11 @@ public class Directory implements MetadataValue equivalentTo(directory); } - private static Pair minMaxVersions(BTreeMap states, BTreeMap versions) + private static ClusterVersions clusterVersions(BTreeMap states, BTreeMap versions) { NodeVersion minVersion = null; NodeVersion maxVersion = null; + int commonVersion = Integer.MAX_VALUE; for (Map.Entry entry : states.entrySet()) { if (entry.getValue() != NodeState.LEFT) @@ -782,26 +826,15 @@ public class Directory implements MetadataValue minVersion = ver; if (maxVersion == null || ver.compareTo(maxVersion) > 0) maxVersion = ver; - } - } - if (minVersion == null) - return Pair.create(CURRENT, CURRENT); - return Pair.create(minVersion, maxVersion); - } - - public static Version minCommonSerializationVersion(BTreeMap states, BTreeMap versions) - { - int commonVersion = Integer.MAX_VALUE; - for (Map.Entry entry : states.entrySet()) - { - if (entry.getValue() != NodeState.LEFT) - { - NodeVersion ver = versions.get(entry.getKey()); if (ver.serializationVersion > Version.OLD.asInt() && ver.serializationVersion < commonVersion) commonVersion = ver.serializationVersion; } } - return commonVersion == Integer.MAX_VALUE ? NodeVersion.CURRENT_METADATA_VERSION : Version.fromInt(commonVersion); + if (minVersion == null) + return new ClusterVersions(CURRENT, CURRENT, CURRENT_METADATA_VERSION); + + return new ClusterVersions(minVersion, maxVersion, + commonVersion == Integer.MAX_VALUE ? NodeVersion.CURRENT_METADATA_VERSION : Version.fromInt(commonVersion)); } @Override @@ -825,7 +858,8 @@ public class Directory implements MetadataValue Objects.equals(endpointsByDC, directory.endpointsByDC) && Objects.equals(racksByDC, directory.racksByDC) && Objects.equals(versions, directory.versions) && - Objects.equals(addresses, directory.addresses); + Objects.equals(addresses, directory.addresses) && + Objects.equals(removedNodes, directory.removedNodes); } private static final Logger logger = LoggerFactory.getLogger(Directory.class); @@ -891,7 +925,6 @@ public class Directory implements MetadataValue } - public static class RemovedNode implements Comparable { public final Epoch removedIn; @@ -923,4 +956,18 @@ public class Directory implements MetadataValue return id.compareTo(o.id); } } + + private static class ClusterVersions + { + private final NodeVersion clusterMinVersion; + private final NodeVersion clusterMaxVersion; + private final Version commonSerializationVersion; + public ClusterVersions(NodeVersion clusterMinVersion, NodeVersion clusterMaxVersion, Version commonSerializationVersion) + { + + this.clusterMinVersion = clusterMinVersion; + this.clusterMaxVersion = clusterMaxVersion; + this.commonSerializationVersion = commonSerializationVersion; + } + } } diff --git a/test/microbench/org/apache/cassandra/test/microbench/DirectorySerializationBench.java b/test/microbench/org/apache/cassandra/test/microbench/DirectorySerializationBench.java new file mode 100644 index 0000000000..5b9bf7bd69 --- /dev/null +++ b/test/microbench/org/apache/cassandra/test/microbench/DirectorySerializationBench.java @@ -0,0 +1,126 @@ +/* + * 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.test.microbench; + +import java.io.IOException; +import java.net.UnknownHostException; +import java.util.Collections; +import java.util.Random; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.io.util.DataInputBuffer; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.RegistrationStatus; +import org.apache.cassandra.tcm.membership.Location; +import org.apache.cassandra.tcm.membership.NodeAddresses; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.membership.NodeVersion; +import org.apache.cassandra.tcm.ownership.PlacementProvider; +import org.apache.cassandra.tcm.ownership.UniformRangePlacement; +import org.apache.cassandra.tcm.transformations.UnsafeJoin; +import org.apache.cassandra.utils.CassandraVersion; + +@State(Scope.Benchmark) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Fork(value = 1) +@Warmup(iterations = 5, timeUnit = TimeUnit.MILLISECONDS, time = 5000) +@Measurement(iterations = 5, timeUnit = TimeUnit.MILLISECONDS, time = 30000) +public class DirectorySerializationBench +{ + static Random random = new Random(1); + static ClusterMetadata metadata; + static byte[] serialized; + @Setup(Level.Trial) + public void setup() throws IOException + { + DatabaseDescriptor.daemonInitialization(); + int nodecount = 4000; + metadata = fakeMetadata(nodecount, 3, 3); + RegistrationStatus.instance.onRegistration(); + DataOutputBuffer buf = new DataOutputBuffer(2_000_000); + ClusterMetadata.serializer.serialize(metadata, buf, NodeVersion.CURRENT_METADATA_VERSION); + serialized = buf.toByteArray(); + } + + @Benchmark + public void bench() throws IOException + { + ClusterMetadata.serializer.deserialize(new DataInputBuffer(serialized), NodeVersion.CURRENT_METADATA_VERSION); + } + + public static ClusterMetadata fakeMetadata(int nodeCount, int dcCount, int rackCount) throws UnknownHostException + { + ClusterMetadata metadata = new ClusterMetadata(Murmur3Partitioner.instance); + TokenSupplier tokensupplier = TokenSupplier.evenlyDistributedTokens(nodeCount); + PlacementProvider placementProvider = new UniformRangePlacement(); + NodeVersion nodeVersion = new NodeVersion(new CassandraVersion("6.0.0"), NodeVersion.CURRENT_METADATA_VERSION); + for (int i = 1; i < nodeCount; i++) + { + ClusterMetadata.Transformer transformer = metadata.transformer(); + UUID uuid = UUID.randomUUID(); + NodeAddresses addresses = addresses(uuid, i); + metadata = transformer.register(addresses, new Location("dc" + random.nextInt(dcCount), "rack"+random.nextInt(rackCount)), nodeVersion).build().metadata; + NodeId nodeId = metadata.directory.peerId(addresses.broadcastAddress); + metadata = new UnsafeJoin(nodeId, Collections.singleton(new Murmur3Partitioner.LongToken(tokensupplier.token(i))), placementProvider).execute(metadata).success().metadata; + } + + return metadata; + } + + static NodeAddresses addresses(UUID uuid, int idx) throws UnknownHostException + { + byte [] address = new byte [] {127, 0, + (byte) (((idx + 1) & 0x0000ff00) >> 8), + (byte) ((idx + 1) & 0x000000ff)}; + + InetAddressAndPort host = InetAddressAndPort.getByAddress(address); + return new NodeAddresses(uuid, host, host, host); + } + + public static void main(String[] args) throws RunnerException, UnknownHostException + { + Options options = new OptionsBuilder() + .include(DirectorySerializationBench.class.getSimpleName()) + .build(); + new Runner(options).run(); + } +} From 740879d5a046768be0fd6c171457dad411d05626 Mon Sep 17 00:00:00 2001 From: Marcus Eriksson Date: Mon, 16 Mar 2026 15:04:42 +0100 Subject: [PATCH 4/6] Don't clear prepared statement cache on nodetool cms initialize Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-21234 --- CHANGES.txt | 1 + .../apache/cassandra/tcm/ClusterMetadata.java | 5 - .../tcm/listeners/InitializationListener.java | 36 ------ .../apache/cassandra/tcm/log/LocalLog.java | 2 - ...MetadataUpgradePreparedStatementsTest.java | 110 ++++++++++++++++++ 5 files changed, 111 insertions(+), 43 deletions(-) delete mode 100644 src/java/org/apache/cassandra/tcm/listeners/InitializationListener.java create mode 100644 test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePreparedStatementsTest.java diff --git a/CHANGES.txt b/CHANGES.txt index ff76806971..8d1e71aec1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234) * Improve performance when deserializing cluster metadata (CASSANDRA-21224) * Minor TokenMap performance improvement (CASSANDRA-21223) * Handle lost response when committing PrepareMove (CASSANDRA-21222) diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java index b99663a655..5645b439e4 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -1062,11 +1062,6 @@ public class ClusterMetadata return null; } - public boolean metadataSerializationUpgradeInProgress() - { - return !directory.clusterMaxVersion.serializationVersion().equals(directory.commonSerializationVersion); - } - public static class Serializer implements MetadataSerializer { @Override diff --git a/src/java/org/apache/cassandra/tcm/listeners/InitializationListener.java b/src/java/org/apache/cassandra/tcm/listeners/InitializationListener.java deleted file mode 100644 index 5ce601d46e..0000000000 --- a/src/java/org/apache/cassandra/tcm/listeners/InitializationListener.java +++ /dev/null @@ -1,36 +0,0 @@ -/* - * 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.listeners; - -import org.apache.cassandra.cql3.QueryProcessor; -import org.apache.cassandra.tcm.Transformation; -import org.apache.cassandra.tcm.log.Entry; - -public class InitializationListener implements LogListener -{ - @Override - public void notify(Entry entry, Transformation.Result result) - { - if (entry.transform.kind() == Transformation.Kind.INITIALIZE_CMS) - { - QueryProcessor.clearPreparedStatementsCache(); - QueryProcessor.clearInternalStatementsCache(); - } - } -} diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java index 4b7760e002..b3069eb029 100644 --- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java +++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java @@ -57,7 +57,6 @@ import org.apache.cassandra.tcm.Startup; import org.apache.cassandra.tcm.Transformation; import org.apache.cassandra.tcm.listeners.ChangeListener; import org.apache.cassandra.tcm.listeners.ClientNotificationListener; -import org.apache.cassandra.tcm.listeners.InitializationListener; import org.apache.cassandra.tcm.listeners.LegacyStateListener; import org.apache.cassandra.tcm.listeners.LogListener; import org.apache.cassandra.tcm.listeners.MetadataSnapshotListener; @@ -905,7 +904,6 @@ public abstract class LocalLog implements Closeable changeListeners.clear(); addListener(snapshotListener()); - addListener(new InitializationListener()); addListener(new SchemaListener(spec.loadSSTables)); addListener(new LegacyStateListener()); addListener(new PlacementsChangeListener()); diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePreparedStatementsTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePreparedStatementsTest.java new file mode 100644 index 0000000000..81b48040c5 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePreparedStatementsTest.java @@ -0,0 +1,110 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.upgrade; + +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.Session; + +import org.junit.Test; + +import org.apache.cassandra.cql3.QueryHandler; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.statements.SelectStatement; +import org.apache.cassandra.cql3.statements.UpdateStatement; +import org.apache.cassandra.distributed.Constants; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.IUpgradeableInstance; + +public class ClusterMetadataUpgradePreparedStatementsTest extends UpgradeTestBase +{ + @Test + public void simpleUpgradeTest() throws Throwable + { + new UpgradeTestBase.TestCase() + .nodes(3) + .nodesToUpgrade(1, 2, 3) + .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP, Feature.NATIVE_PROTOCOL) + .set(Constants.KEY_DTEST_FULL_STARTUP, true)) + .upgradesToCurrentFrom(v50) + .setup((cluster) -> { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))")); + }) + .runAfterClusterUpgrade((cluster) -> { + String node1Address = cluster.get(1).config().broadcastAddress().getHostString(); + com.datastax.driver.core.Cluster.Builder builder = com.datastax.driver.core.Cluster.builder() + .addContactPoint(node1Address); + try (com.datastax.driver.core.Cluster c = builder.build(); + Session session = c.connect()) + { + PreparedStatement ps = session.prepare(withKeyspace("insert into %s.tbl (pk, ck, v) values (?, ?, ?)")); + PreparedStatement ps2 = session.prepare(withKeyspace("select pk, ck, v from %s.tbl where pk = ?")); + for (int i = 0; i < 10; i++) + { + session.execute(ps.bind(i, 2, 3)); + session.execute(ps2.bind(i)); + } + assertPSCache(cluster.get(1), false, 0); + cluster.get(1).nodetoolResult("cms", "initialize").asserts().success(); + assertPSCache(cluster.get(1), false, 0); + for (int i = 0; i < 10; i++) + { + session.execute(ps.bind(i, 3, 3)); + session.execute(ps2.bind(i)); + } + session.execute(withKeyspace("alter table %s.tbl add x int")); + + assertPSCacheEmpty(cluster.get(1)); + + for (int i = 0; i < 10; i++) + { + session.execute(ps.bind(i, 3, 3)); + session.execute(ps2.bind(i)); + } + assertPSCache(cluster.get(1), false, 4); + } + }).run(); + } + + private static void assertPSCacheEmpty(IUpgradeableInstance inst) + { + assertPSCache(inst, true, -1); + } + + private static void assertPSCache(IUpgradeableInstance inst, boolean shouldBeEmpty, long expectedEpoch) + { + ((IInvokableInstance)inst).runOnInstance(() -> { + if (shouldBeEmpty != QueryProcessor.instance.getPreparedStatements().isEmpty()) + throw new AssertionError("Prepared statements should not be empty after `cms initialize`"); + + for (QueryHandler.Prepared p : QueryProcessor.instance.getPreparedStatements().values()) + { + long statementEpoch = -1; + if (p.statement instanceof SelectStatement) + statementEpoch = ((SelectStatement)p.statement).table.epoch.getEpoch(); + else if (p.statement instanceof UpdateStatement) + statementEpoch = ((UpdateStatement)p.statement).metadata.epoch.getEpoch(); + + if (statementEpoch != expectedEpoch) + throw new AssertionError(String.format("Statement %s has the wrong epoch: %d != %d ", p.statement, statementEpoch, expectedEpoch)); + } + }); + } + +} From ea495907e136c43dcc4d60ab6694f2d831dae325 Mon Sep 17 00:00:00 2001 From: Marcus Eriksson Date: Mon, 16 Mar 2026 15:04:42 +0100 Subject: [PATCH 5/6] Make nodetool abortbootstrap more robust Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-21235 --- CHANGES.txt | 1 + .../org/apache/cassandra/gms/Gossiper.java | 8 +- .../cassandra/service/StorageService.java | 55 +++++++-- .../tcm/transformations/Register.java | 3 + .../distributed/test/ring/BootstrapTest.java | 24 +++- ...terMetadataAbortedBootstrapRejoinTest.java | 109 ++++++++++++++++++ 6 files changed, 190 insertions(+), 10 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/tcm/ClusterMetadataAbortedBootstrapRejoinTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 8d1e71aec1..698c97c150 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Make nodetool abortbootstrap more robust (CASSANDRA-21235) * Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234) * Improve performance when deserializing cluster metadata (CASSANDRA-21224) * Minor TokenMap performance improvement (CASSANDRA-21223) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 728ae7de7c..3d6414913f 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -269,7 +269,13 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, taskLock.lock(); /* Update the local heartbeat counter. */ - endpointStateMap.get(getBroadcastAddressAndPort()).updateHeartBeat(); + EndpointState epstate = endpointStateMap.get(getBroadcastAddressAndPort()); + if (epstate == null) + { + logger.warn("Node {} is not in gossip, not running GossipTask", getBroadcastAddressAndPort()); + return; + } + epstate.updateHeartBeat(); if (logger.isTraceEnabled()) logger.trace("My heartbeat is now {}", endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().getHeartBeatVersion()); final List gDigests = new ArrayList<>(); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index f99599c889..d32d19460d 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -1662,15 +1662,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public void abortBootstrap(String nodeStr, String endpointStr) { - logger.debug("Aborting bootstrap for {}/{}", nodeStr, endpointStr); + logger.info("Aborting bootstrap for {}", StringUtils.isEmpty(nodeStr) ? endpointStr : nodeStr); ClusterMetadata metadata = ClusterMetadata.current(); - NodeId nodeId; - if (!StringUtils.isEmpty(nodeStr)) - nodeId = NodeId.fromString(nodeStr); - else - nodeId = metadata.directory.peerId(InetAddressAndPort.getByNameUnchecked(endpointStr)); - + NodeId nodeId = parseNodeIdOrEndpoint(metadata, nodeStr, endpointStr); InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId); + if (endpoint == null) + throw new IllegalArgumentException("Can't abort bootstrap for " + nodeId + " - it does not exist in cluster metadata"); if (Gossiper.instance.isKnownEndpoint(endpoint) && FailureDetector.instance.isAlive(endpoint)) throw new RuntimeException("Can't abort bootstrap for " + nodeId + " - it is alive"); NodeState nodeState = metadata.directory.peerState(nodeId); @@ -1693,6 +1690,47 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } + private static NodeId parseNodeIdOrEndpoint(ClusterMetadata metadata, String nodeStr, String endpointStr) + { + NodeId nodeId; + if (!StringUtils.isEmpty(nodeStr)) + { + try + { + nodeId = NodeId.fromString(nodeStr); + } + catch (IllegalArgumentException | UnsupportedOperationException e) + { + String msg = "Unable to parse node id string " + nodeStr; + logger.warn("{}", msg, e); + throw new IllegalArgumentException(msg, e); + } + } + else + { + InetAddressAndPort endpoint; + try + { + endpoint = InetAddressAndPort.getByName(endpointStr); + } + catch (UnknownHostException e) + { + String msg = "Unable to look up endpoint " + endpointStr; + logger.warn("{}", msg, e); + throw new IllegalArgumentException(msg, e); + } + + nodeId = metadata.directory.peerId(endpoint); + if (nodeId == null) + { + String msg = "Unknown endpoint: " + endpoint; + logger.warn(msg); + throw new IllegalArgumentException(msg); + } + } + return nodeId; + } + @Override public void migrateConsensusProtocol(@Nonnull List keyspaceNames, @Nullable List maybeTableNames, @@ -2009,7 +2047,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public String getLocalHostId() { - return getLocalHostUUID().toString(); + UUID localHostId = getLocalHostUUID(); + return localHostId != null ? localHostId.toString() : "UNKNOWN"; } public UUID getLocalHostUUID() diff --git a/src/java/org/apache/cassandra/tcm/transformations/Register.java b/src/java/org/apache/cassandra/tcm/transformations/Register.java index 1e31ff3a85..a7551c7069 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/Register.java +++ b/src/java/org/apache/cassandra/tcm/transformations/Register.java @@ -176,6 +176,9 @@ public class Register implements Transformation else { NodeId nodeId = directory.peerId(FBUtilities.getBroadcastAddressAndPort()); + if (nodeId == null) + throw new IllegalStateException("Node has host id "+localHostId+" in system.local, but is not present in cluster metadata - not allowing this node to register. " + + "If a bootstrap of this node failed and was aborted with `nodetool abortbootstrap` it should also have its data removed before trying to rebootstrap."); NodeVersion dirVersion = directory.version(nodeId); // If this is a node in the process of upgrading, update the host id in the system.local table diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java index bc7abb85ef..8012399afd 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java @@ -18,6 +18,8 @@ package org.apache.cassandra.distributed.test.ring; +import java.io.Closeable; +import java.io.IOException; import java.util.Map; import java.util.Objects; import java.util.concurrent.Callable; @@ -48,6 +50,7 @@ import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.ICluster; 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.JMXUtil; import org.apache.cassandra.distributed.shared.NetworkTopology; @@ -55,7 +58,6 @@ import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.metrics.DefaultNameFactory; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.StorageServiceMBean; -import org.apache.cassandra.utils.Closeable; import org.apache.cassandra.utils.concurrent.CountDownLatch; import static net.bytebuddy.matcher.ElementMatchers.named; @@ -228,6 +230,26 @@ public class BootstrapTest extends TestBaseImpl } } + @Test + public void testAbortBootstrapBadIp() throws IOException + { + try (Cluster cluster = builder().withNodes(1).start()) + { + NodeToolResult res = cluster.get(1).nodetoolResult("abortbootstrap", "--ip", "127.0.0.55"); + res.asserts().failure(); + assertTrue(res.getStdout().contains("Unknown endpoint")); + res = cluster.get(1).nodetoolResult("abortbootstrap", "--ip", "127.0.0.999"); + res.asserts().failure(); + assertTrue(res.getStdout().contains("Unable to look up endpoint")); + res = cluster.get(1).nodetoolResult("abortbootstrap", "--node", "999"); + res.asserts().failure(); + assertTrue(res.getStdout().contains("does not exist in cluster metadata")); + res = cluster.get(1).nodetoolResult("abortbootstrap", "--node", "hello"); + res.asserts().failure(); + assertTrue(res.getStdout().contains("Unable to parse node id")); + } + } + public static void populate(ICluster cluster, int from, int to) { populate(cluster, from, to, 1, 3, ConsistencyLevel.QUORUM); diff --git a/test/distributed/org/apache/cassandra/distributed/test/tcm/ClusterMetadataAbortedBootstrapRejoinTest.java b/test/distributed/org/apache/cassandra/distributed/test/tcm/ClusterMetadataAbortedBootstrapRejoinTest.java new file mode 100644 index 0000000000..5994fab07c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/tcm/ClusterMetadataAbortedBootstrapRejoinTest.java @@ -0,0 +1,109 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test.tcm; + +import java.io.IOException; +import java.time.Duration; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; + +import com.google.common.util.concurrent.Uninterruptibles; + +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; + +import org.assertj.core.api.Assertions; +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.service.StorageService; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + + +public class ClusterMetadataAbortedBootstrapRejoinTest extends TestBaseImpl +{ + @Test + public void testFailedBootstrapNotAllowedToJoin() throws IOException, TimeoutException, ExecutionException, InterruptedException + { + TokenSupplier even = TokenSupplier.evenlyDistributedTokens(3); + try (Cluster cluster = init(Cluster.build(2) + .withInstanceInitializer(BBHelper::install) + .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK)) + .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(3, "dc0", "rack0")) + .withTokenSupplier(even::token) + .start())) + { + IInstanceConfig config = cluster.newInstanceConfig() + .set("auto_bootstrap", true); + IInvokableInstance toBootstrap = cluster.bootstrap(config); + toBootstrap.startup(cluster); + toBootstrap.logs().watchFor(Duration.ofSeconds(60), BBHelper.FAILMESSAGE); + toBootstrap.shutdown().get(); + cluster.get(1).runOnInstance(() -> { + int i = 0; + while (FailureDetector.instance.isAlive(InetAddressAndPort.getByNameUnchecked("127.0.0.3")) && i++ < 30) + Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); + }); + cluster.get(1).nodetoolResult("abortbootstrap", "--ip", "127.0.0.3").asserts().success(); + Assertions.assertThatThrownBy(toBootstrap::startup) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("but is not present in cluster metadata"); + } + } + + public static class BBHelper + { + public static String FAILMESSAGE = "ARTIFICIALLY FAILING BOOTSTRAP"; + public static AtomicBoolean enabled = new AtomicBoolean(true); + public static void install(ClassLoader cl, int i) + { + if (i == 3) + { + new ByteBuddy().rebase(StorageService.class) + .method(named("repairPaxosForTopologyChange").and(takesArguments(1))) + .intercept(MethodDelegation.to(BBHelper.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + } + + public static void repairPaxosForTopologyChange(String reason) + { + if (enabled.get()) + { + enabled.set(false); + throw new RuntimeException(FAILMESSAGE); + } + } + + } +} From 8fa9a75fc2e6eae26e40fc5df3636fcf2f93d1bc Mon Sep 17 00:00:00 2001 From: Marcus Eriksson Date: Mon, 16 Mar 2026 15:04:42 +0100 Subject: [PATCH 6/6] =?UTF-8?q?Don=E2=80=99t=20leave=20autocompaction=20di?= =?UTF-8?q?sabled=20during=20bootstrap=20and=20replace?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-21236 --- CHANGES.txt | 1 + .../cassandra/service/CassandraDaemon.java | 10 +- .../test/ring/BootstrapCompactionTest.java | 93 +++++++++++++++++++ .../CompactionStrategyManagerTest.java | 21 +++-- 4 files changed, 113 insertions(+), 12 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapCompactionTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 698c97c150..628ba1f48f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * Don’t leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236) * Make nodetool abortbootstrap more robust (CASSANDRA-21235) * Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234) * Improve performance when deserializing cluster metadata (CASSANDRA-21224) diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 2825041f96..a69124bb7d 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -392,6 +392,11 @@ public class CassandraDaemon // Prepared statements QueryProcessor.instance.preloadPreparedStatements(); + // Apply overrides before re-enabling auto-compaction + setCompactionStrategyOverrides(Schema.instance.getKeyspaces()); + // re-enable auto-compaction after replay, so correct disk boundaries are used + enableAutoCompaction(Schema.instance.getKeyspaces()); + // start server internals StorageService.instance.registerDaemon(this); try @@ -424,11 +429,6 @@ public class CassandraDaemon ScheduledExecutors.optionalTasks.schedule(viewRebuild, StorageService.RING_DELAY_MILLIS, TimeUnit.MILLISECONDS); StorageService.instance.doAuthSetup(); - // Apply overrides before re-enabling auto-compaction - setCompactionStrategyOverrides(Schema.instance.getKeyspaces()); - // re-enable auto-compaction after replay, so correct disk boundaries are used - enableAutoCompaction(Schema.instance.getKeyspaces()); - AuditLogManager.instance.initialize(); StorageService.instance.doAutoRepairSetup(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapCompactionTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapCompactionTest.java new file mode 100644 index 0000000000..e53e3cef86 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapCompactionTest.java @@ -0,0 +1,93 @@ +/* + * 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.ring; + +import java.util.concurrent.Callable; + +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; +import net.bytebuddy.implementation.bind.annotation.SuperCall; + +import org.junit.Test; + +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.Constants; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.tcm.sequences.BootstrapAndJoin; +import org.apache.cassandra.tcm.sequences.SequenceState; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.junit.Assert.assertTrue; + +public class BootstrapCompactionTest extends TestBaseImpl +{ + @Test + public void testCompactionEnabledDuringBootstrap() throws Exception + { + int originalNodeCount = 2; + int expandedNodeCount = originalNodeCount + 1; + + try (Cluster cluster = init(builder().withNodes(originalNodeCount) + .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(expandedNodeCount)) + .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0")) + .withInstanceInitializer(BB::install) + .withConfig(config -> config.with(NETWORK, GOSSIP)) + .start())) + { + cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key)")); + IInstanceConfig config = cluster.newInstanceConfig() + .set(Constants.KEY_DTEST_FULL_STARTUP, true) + .set("auto_bootstrap", true); + + IInvokableInstance newInstance = cluster.bootstrap(config); + // BB below asserts that autocompaction is enabled at each step in the join sequence + newInstance.startup(cluster); + } + } + + public static class BB + { + public static void install(ClassLoader cl, int i) + { + if (i == 3) + { + new ByteBuddy().rebase(BootstrapAndJoin.class) + .method(named("executeNext")) + .intercept(MethodDelegation.to(BB.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + } + + public static SequenceState executeNext(@SuperCall Callable zuper) throws Exception + { + boolean isEnabled = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").getCompactionStrategyManager().isEnabled(); + assertTrue("Autocompaction should be enabled during the bootstrap", isEnabled); + return zuper.call(); + } + } +} diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionStrategyManagerTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionStrategyManagerTest.java index b52b550187..5e744f4758 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionStrategyManagerTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionStrategyManagerTest.java @@ -214,8 +214,9 @@ public class CompactionStrategyManagerTest extends CassandraTestBase // inside the currentlyBackgroundUpgrading check - with max_concurrent_auto_upgrade_tasks = 1 this will make // sure that BackgroundCompactionCandidate#maybeRunUpgradeTask returns false until the latch has been counted down CountDownLatch latch = new CountDownLatch(1); + CountDownLatch inFindUpgradeSSTables = new CountDownLatch(1); AtomicInteger upgradeTaskCount = new AtomicInteger(0); - MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, upgradeTaskCount); + MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, inFindUpgradeSSTables, upgradeTaskCount); CompactionManager.BackgroundCompactionCandidate r = CompactionManager.instance.getBackgroundCompactionCandidate(mock); CompactionStrategyManager mgr = mock.getCompactionStrategyManager(); @@ -224,7 +225,7 @@ public class CompactionStrategyManagerTest extends CassandraTestBase // due to the currentlyBackgroundUpgrading count being >= max_concurrent_auto_upgrade_tasks Thread t = new Thread(() -> r.maybeRunUpgradeTask(mgr)); t.start(); - Thread.sleep(100); // let the thread start and grab the task + inFindUpgradeSSTables.await(); assertEquals(1, CompactionManager.instance.currentlyBackgroundUpgrading.get()); assertFalse(r.maybeRunUpgradeTask(mgr)); assertFalse(r.maybeRunUpgradeTask(mgr)); @@ -246,8 +247,9 @@ public class CompactionStrategyManagerTest extends CassandraTestBase // inside the currentlyBackgroundUpgrading check - with max_concurrent_auto_upgrade_tasks = 1 this will make // sure that BackgroundCompactionCandidate#maybeRunUpgradeTask returns false until the latch has been counted down CountDownLatch latch = new CountDownLatch(1); + CountDownLatch inFindUpgradeSSTables = new CountDownLatch(2); AtomicInteger upgradeTaskCount = new AtomicInteger(); - MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, upgradeTaskCount); + MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, inFindUpgradeSSTables, upgradeTaskCount); CompactionManager.BackgroundCompactionCandidate r = CompactionManager.instance.getBackgroundCompactionCandidate(mock); CompactionStrategyManager mgr = mock.getCompactionStrategyManager(); @@ -259,7 +261,7 @@ public class CompactionStrategyManagerTest extends CassandraTestBase t.start(); Thread t2 = new Thread(() -> r.maybeRunUpgradeTask(mgr)); t2.start(); - Thread.sleep(100); // let the threads start and grab the task + inFindUpgradeSSTables.await(); assertEquals(2, CompactionManager.instance.currentlyBackgroundUpgrading.get()); assertFalse(r.maybeRunUpgradeTask(mgr)); assertFalse(r.maybeRunUpgradeTask(mgr)); @@ -619,18 +621,20 @@ public class CompactionStrategyManagerTest extends CassandraTestBase private static class MockCFSForCSM extends ColumnFamilyStore { private final CountDownLatch latch; + private final CountDownLatch inFindUpgradeSSTables; private final AtomicInteger upgradeTaskCount; - private MockCFSForCSM(ColumnFamilyStore cfs, CountDownLatch latch, AtomicInteger upgradeTaskCount) + private MockCFSForCSM(ColumnFamilyStore cfs, CountDownLatch latch, CountDownLatch inFindUpgradeSSTables, AtomicInteger upgradeTaskCount) { super(cfs.keyspace, cfs.name, Util.newSeqGen(10), cfs.metadata.get(), cfs.getDirectories(), true, false); this.latch = latch; + this.inFindUpgradeSSTables = inFindUpgradeSSTables; this.upgradeTaskCount = upgradeTaskCount; } @Override public CompactionStrategyManager getCompactionStrategyManager() { - return new MockCSM(this, latch, upgradeTaskCount); + return new MockCSM(this, latch, inFindUpgradeSSTables, upgradeTaskCount); } } @@ -638,11 +642,13 @@ public class CompactionStrategyManagerTest extends CassandraTestBase { private final CountDownLatch latch; private final AtomicInteger upgradeTaskCount; + private final CountDownLatch inFindUpgradeSSTables; - private MockCSM(ColumnFamilyStore cfs, CountDownLatch latch, AtomicInteger upgradeTaskCount) + private MockCSM(ColumnFamilyStore cfs, CountDownLatch latch, CountDownLatch inFindUpgradeSSTables, AtomicInteger upgradeTaskCount) { super(cfs); this.latch = latch; + this.inFindUpgradeSSTables = inFindUpgradeSSTables; this.upgradeTaskCount = upgradeTaskCount; } @@ -651,6 +657,7 @@ public class CompactionStrategyManagerTest extends CassandraTestBase { try { + inFindUpgradeSSTables.countDown(); latch.await(); upgradeTaskCount.incrementAndGet(); }