You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2026/08/13 22:36:08 UTC
(cassandra) branch cassandra-6.0 updated (9071b403cd -> a2376d9127)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch cassandra-6.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
from 9071b403cd Merge branch 'cassandra-5.0' into cassandra-6.0
new 24dfcfe057 Ensure transferred_ranges reset on decommision re-attempt when pending ranges cannot be proven continous
new 8fa203de28 Merge branch 'cassandra-4.0' into cassandra-4.1
new 6e6077ebba Merge branch 'cassandra-4.1' into cassandra-5.0
new a2376d9127 Merge branch 'cassandra-5.0' into cassandra-6.0
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 1 +
.../org/apache/cassandra/db/SystemKeyspace.java | 31 ++++++-
.../cassandra/tcm/sequences/RemoveNodeStreams.java | 2 +-
.../tcm/sequences/SingleNodeSequences.java | 5 ++
.../tcm/sequences/UnbootstrapStreams.java | 5 +-
.../distributed/test/ring/BootstrapTest.java | 7 +-
.../distributed/test/ring/DecommissionTest.java | 96 +++++++++++++++++++++-
7 files changed, 139 insertions(+), 8 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/01: Merge branch 'cassandra-5.0' into cassandra-6.0
Posted by ma...@apache.org.
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
commit a2376d9127db980f2fcb63bd11fd4777816e175f
Merge: 9071b403cd 6e6077ebba
Author: Caleb Rackliffe <ca...@gmail.com>
AuthorDate: Thu Aug 13 17:21:08 2026 -0500
Merge branch 'cassandra-5.0' into cassandra-6.0
* cassandra-5.0:
Ensure transferred_ranges reset on decommision re-attempt when pending ranges cannot be proven continous
CHANGES.txt | 1 +
.../org/apache/cassandra/db/SystemKeyspace.java | 31 ++++++-
.../cassandra/tcm/sequences/RemoveNodeStreams.java | 2 +-
.../tcm/sequences/SingleNodeSequences.java | 5 ++
.../tcm/sequences/UnbootstrapStreams.java | 5 +-
.../distributed/test/ring/BootstrapTest.java | 7 +-
.../distributed/test/ring/DecommissionTest.java | 96 +++++++++++++++++++++-
7 files changed, 139 insertions(+), 8 deletions(-)
diff --cc CHANGES.txt
index ddbb8c5145,8d2e5db0fb..bbfb9c2ba8
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,16 -1,10 +1,17 @@@
-5.0.10
+6.0-alpha3
++ * Ensure transferred_ranges reset on decommision re-attempt when pending ranges cannot be proven continous (CASSANDRA-16290)
+ * Add blob type support to SAI (CASSANDRA-20012)
+ * Fix deserialization of column masks in cluster metadata (CASSANDRA-21549)
+ * Caffeine caches in CompressionDictionaryCache and ZstdDictionaryCompressor should specify an executor explicitly (CASSANDRA-21557)
+ * Make cqlsh prompt to reset to no keyspace set by USE after dropping that keyspace (CASSANDRA-21548)
+ * Implement CMS rediscovery and recovery protocol (CASSANDRA-20476)
+Merged from 5.0:
* Avoid rebuilding per-SSTable SAI components unless missing or corrupted (CASSANDRA-21515)
* Propagate trickle_fsync settings to compressed SSTable writers (CASSANDRA-21487)
- * Allow DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522)
- * Return CorruptSSTableException if chunk metadata and file size are out of sync (CASSANDRA-21519)
+ * Allow setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522)
+ * Fix ThreadLocalReadAheadBuffer#fill() to throw a CorruptBlockException if chunk metadata and file size are out of sync (CASSANDRA-21519)
+ * Fix memtable on-heap accounting drift in BTree.update and BTreeRow.merge (CASSANDRA-21472)
Merged from 4.0:
- * Ensure transferred_ranges reset on decommision re-attempt when pending ranges cannot be proven continous (CASSANDRA-16290)
* Add validation to uncompressed length during decompression (CASSANDRA-21567)
* Fix regression in PasswordObfuscator for dollar-quoted passwords (CASSANDRA-21559)
* Do not make DNS lookup when querying system_views.clients for hostname column by removing it (CASSANDRA-21539)
diff --cc src/java/org/apache/cassandra/tcm/sequences/RemoveNodeStreams.java
index bd09f8aaca,0000000000..dbaaef7382
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/tcm/sequences/RemoveNodeStreams.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/RemoveNodeStreams.java
@@@ -1,142 -1,0 +1,142 @@@
+/*
+ * 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.sequences;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.locator.EndpointsByReplica;
+import org.apache.cassandra.locator.EndpointsForRange;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.RangesByEndpoint;
+import org.apache.cassandra.locator.ReplicaCollection;
+import org.apache.cassandra.locator.SystemStrategy;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.streaming.DataMovement;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.ownership.MovementMap;
+import org.apache.cassandra.tcm.ownership.PlacementDeltas;
+import org.apache.cassandra.tcm.ownership.ReplicaGroups;
+
+import static org.apache.cassandra.streaming.StreamOperation.RESTORE_REPLICA_COUNT;
+
+public class RemoveNodeStreams implements LeaveStreams
+{
- private static final Logger logger = LoggerFactory.getLogger(UnbootstrapStreams.class);
++ private static final Logger logger = LoggerFactory.getLogger(RemoveNodeStreams.class);
+ private final AtomicBoolean finished = new AtomicBoolean();
+ private final AtomicBoolean failed = new AtomicBoolean();
+ private DataMovements.ResponseTracker responseTracker;
+
+ @Override
+ public void execute(NodeId leaving, PlacementDeltas startLeave, PlacementDeltas midLeave, PlacementDeltas finishLeave) throws ExecutionException, InterruptedException
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ MovementMap movements = movementMap(metadata.directory.endpoint(leaving),
+ metadata,
+ startLeave);
+ movements.forEach((params, eps) -> logger.info("Removenode movements: {}: {}", params, eps));
+ String operationId = leaving.toUUID().toString();
+ responseTracker = DataMovements.instance.registerMovements(RESTORE_REPLICA_COUNT, operationId, movements);
+ movements.byEndpoint().forEach((endpoint, epMovements) -> {
+ DataMovement msg = new DataMovement(operationId, RESTORE_REPLICA_COUNT.name(), epMovements);
+ MessagingService.instance().sendWithCallback(Message.out(Verb.INITIATE_DATA_MOVEMENTS_REQ, msg), endpoint, response -> {
+ logger.debug("Endpoint {} starting streams {}", response.from(), epMovements);
+ });
+ });
+
+ try
+ {
+ responseTracker.await();
+ finished.set(true);
+ }
+ catch (Exception e)
+ {
+ failed.set(true);
+ throw e;
+ }
+ finally
+ {
+ DataMovements.instance.unregisterMovements(RESTORE_REPLICA_COUNT, operationId);
+ }
+ }
+
+ @Override
+ public Kind kind()
+ {
+ return Kind.REMOVENODE;
+ }
+
+ public String status()
+ {
+ if (finished.get())
+ return "streaming finished";
+ if (failed.get())
+ return "streaming failed";
+ if (responseTracker == null)
+ return "streaming not yet started";
+ return responseTracker.remaining()
+ .stream()
+ .map(i -> i.toString(true))
+ .collect(Collectors.joining(",", "Waiting on streaming responses from: ", ""));
+ }
+
+ /**
+ * create a map where the key is the destination, and the values are possible sources
+ * @return
+ */
+ private static MovementMap movementMap(InetAddressAndPort leaving, ClusterMetadata metadata, PlacementDeltas startDelta)
+ {
+ MovementMap.Builder allMovements = MovementMap.builder();
+ // map of dest->src* movements, keyed by replication settings. During unbootstrap, this will be used to construct
+ // a stream plan for each keyspace, based on their replication params.
+ startDelta.forEach((params, delta) -> {
+ // no streaming for LocalStrategy and friends
+ if (SystemStrategy.class.isAssignableFrom(params.klass))
+ return;
+
+ EndpointsByReplica.Builder movements = new EndpointsByReplica.Builder();
+ RangesByEndpoint startWriteAdditions = startDelta.get(params).writes.additions;
+ RangesByEndpoint startWriteRemovals = startDelta.get(params).writes.removals;
+ // find current placements from the metadata, we need to stream from replicas that are not changed and are therefore not in the deltas
+ ReplicaGroups currentPlacements = metadata.placement(params).reads;
+ startWriteAdditions.flattenValues()
+ .forEach(newReplica -> {
+ EndpointsForRange.Builder candidateBuilder = new EndpointsForRange.Builder(newReplica.range());
+ currentPlacements.forRange(newReplica.range()).get().forEach(replica -> {
+ if (!replica.endpoint().equals(leaving) && !replica.endpoint().equals(newReplica.endpoint()))
+ candidateBuilder.add(replica, ReplicaCollection.Builder.Conflict.NONE);
+ });
+ EndpointsForRange sources = candidateBuilder.build();
+ // log if newReplica is an existing transient replica moving to a full replica
+ if (startWriteRemovals.get(newReplica.endpoint()).contains(newReplica.range(), false))
+ logger.debug("Streaming transient -> full conversion to {} from {}", newReplica.endpoint(), sources);
+ movements.putAll(newReplica, sources, ReplicaCollection.Builder.Conflict.NONE);
+ });
+ allMovements.put(params, movements.build());
+ });
+ return allMovements.build();
+ }
+}
diff --cc src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java
index 82a1059b6e,0000000000..4e3fe9f8e3
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java
@@@ -1,289 -1,0 +1,294 @@@
+/*
+ * 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.sequences;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
+
+import javax.annotation.Nullable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
++import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.MultiStepOperation;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeState;
+import org.apache.cassandra.tcm.transformations.CancelInProgressSequence;
+import org.apache.cassandra.tcm.transformations.PrepareLeave;
+import org.apache.cassandra.tcm.transformations.PrepareMove;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSION_FAILED;
+import static org.apache.cassandra.service.StorageService.Mode.LEAVING;
+import static org.apache.cassandra.service.StorageService.Mode.MOVE_FAILED;
+import static org.apache.cassandra.service.StorageService.Mode.NORMAL;
+import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
+
+/**
+ * This exists simply to group the static entrypoints to sequences that modify a single node
+ * e.g. decommission, remove, move
+ */
+public interface SingleNodeSequences
+{
+ Logger logger = LoggerFactory.getLogger(SingleNodeSequences.class);
+
+ /**
+ * Entrypoint to begin node decommission process.
+ *
+ * @param shutdownNetworking if set to true, will also shut down networking on completion
+ * @param force if set to true, will decommission the node even if this would mean there will be not enough nodes
+ * to satisfy replication factor
+ */
+ static void decommission(boolean shutdownNetworking, boolean force)
+ {
+ if (ClusterMetadataService.instance().isMigrating() || ClusterMetadataService.state() == ClusterMetadataService.State.GOSSIP)
+ throw new IllegalStateException("This cluster is migrating to cluster metadata, can't decommission until that is done.");
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ StorageService.Mode mode = StorageService.instance.operationMode();
+ if (!EnumSet.of(LEAVING, NORMAL, DECOMMISSION_FAILED).contains(mode))
+ throw new UnsupportedOperationException("Node in " + mode + " state; wait for status to become normal");
+ logger.debug("DECOMMISSIONING");
+
+ NodeId self = metadata.myNodeId();
+ Collection<Token> tokens = metadata.tokenMap.tokens(self);
+ ReconfigureCMS.maybeReconfigureCMS(metadata, getBroadcastAddressAndPort());
+ MultiStepOperation<?> inProgress = metadata.inProgressSequences.get(self);
+
+ if (inProgress == null)
+ {
+ logger.info("starting decommission with {} {}", metadata.epoch, self);
++ // We reset transferred ranges upon starting a decommission so that we fully stream
++ // anything written since a previous attempt which may not have been persisted to a pending endpoint
++ SystemKeyspace.resetTransferredRanges();
++ logger.info("done resetting transferred ranges {} {}", metadata.epoch, self);
+ ClusterMetadataService.instance().commit(new PrepareLeave(self,
+ force,
+ ClusterMetadataService.instance().placementProvider(),
+ LeaveStreams.Kind.UNBOOTSTRAP),
+ m -> m,
+ failureHandler("PrepareLeave", StorageService.instance::markDecommissionFailed));
+ }
+ else if (InProgressSequences.isLeave(inProgress))
+ {
+ logger.info("Resuming decommission @ {} (current epoch = {}): {}", inProgress.latestModification, metadata.epoch, inProgress.status());
+ StorageService.instance.clearTransientMode();
+ }
+ else
+ {
+ throw new IllegalArgumentException("Can not decommission a node that has an in-progress sequence");
+ }
+
+ InProgressSequences.finishInProgressSequences(self);
+ Gossiper.instance.unsafeBroadcastLeftStatus(FBUtilities.getBroadcastAddressAndPort(),
+ tokens,
+ metadata.directory.allJoinedEndpoints());
+ if (shutdownNetworking)
+ StorageService.instance.shutdownNetworking();
+ }
+
+ static void abortDecommission(String nodeId)
+ {
+ abortHelper(nodeId, MultiStepOperation.Kind.LEAVE, DECOMMISSION_FAILED);
+ }
+
+ /**
+ * Entrypoint to begin node removal process
+ *
+ * @param toRemove id of the node to remove
+ * @param force if set to true, will remove the node even if this would mean there will be not enough nodes
+ * to satisfy replication factor
+ */
+ static void removeNode(NodeId toRemove, boolean force)
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ if (toRemove.equals(metadata.myNodeId()))
+ throw new UnsupportedOperationException("Cannot remove self");
+ InetAddressAndPort endpoint = metadata.directory.endpoint(toRemove);
+ if (endpoint == null)
+ throw new UnsupportedOperationException("Host ID not found.");
+ if (Gossiper.instance.getLiveMembers().contains(endpoint))
+ throw new UnsupportedOperationException("Node " + endpoint + " is alive and owns this ID. Use decommission command to remove it from the ring");
+
+ NodeState removeState = metadata.directory.peerState(toRemove);
+ if (removeState == null)
+ throw new UnsupportedOperationException("Node to be removed is not a member of the token ring");
+ if (removeState == NodeState.LEAVING)
+ logger.warn("Node {} is already leaving or being removed, continuing removal anyway", endpoint);
+
+ if (metadata.inProgressSequences.contains(toRemove))
+ throw new UnsupportedOperationException("Can not remove a node that has an in-progress sequence");
+
+ ReconfigureCMS.maybeReconfigureCMS(metadata, endpoint);
+
+ logger.info("starting removenode with {} {}", metadata.epoch, toRemove);
+ Collection<Token> tokens = metadata.tokenMap.tokens(toRemove);
+ ClusterMetadataService.instance().commit(new PrepareLeave(toRemove,
+ force,
+ ClusterMetadataService.instance().placementProvider(),
+ LeaveStreams.Kind.REMOVENODE));
+ InProgressSequences.finishInProgressSequences(toRemove);
+ Gossiper.instance.unsafeBroadcastLeftStatus(endpoint, tokens, metadata.directory.allJoinedEndpoints());
+ }
+
+ static void abortRemoveNode(String nodeId)
+ {
+ abortHelper(nodeId, MultiStepOperation.Kind.REMOVE, null);
+ }
+
+ /**
+ * move the node to new token or find a new token to boot to according to load
+ *
+ * @param newToken new token to boot to, or if null, find balanced token to boot to
+ */
+ static void move(Token newToken)
+ {
+ if (ClusterMetadataService.instance().isMigrating() || ClusterMetadataService.state() == ClusterMetadataService.State.GOSSIP)
+ throw new IllegalStateException("This cluster is migrating to cluster metadata, can't move until that is done.");
+
+ if (newToken == null)
+ throw new IllegalArgumentException("Can't move to the undefined (null) token.");
+
+ if (ClusterMetadata.current().tokenMap.tokens().contains(newToken))
+ throw new IllegalArgumentException(String.format("target token %s is already owned by another node.", newToken));
+
+ // address of the current node
+ ClusterMetadata metadata = ClusterMetadata.current();
+ NodeId self = metadata.myNodeId();
+ // This doesn't make any sense in a vnodes environment.
+ if (metadata.tokenMap.tokens(self).size() > 1)
+ {
+ logger.error("Invalid request to move(Token); This node has more than one token and cannot be moved thusly.");
+ throw new UnsupportedOperationException("This node has more than one token and cannot be moved thusly.");
+ }
+
+ ClusterMetadataService.instance().commit(new PrepareMove(self,
+ Collections.singleton(newToken),
+ ClusterMetadataService.instance().placementProvider(),
+ 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<ClusterMetadata> 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)
+ throw new IllegalStateException("This cluster is migrating to cluster metadata, can't move until that is done.");
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+ NodeId self = metadata.myNodeId();
+ MultiStepOperation<?> sequence = metadata.inProgressSequences.get(self);
+ if (sequence == null || sequence.kind() != MultiStepOperation.Kind.MOVE)
+ {
+ 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)
+ {
+ String msg = "Can't resume a move operation unless it has failed";
+ logger.info(msg);
+ throw new IllegalStateException(msg);
+ }
+ StorageService.instance.clearTransientMode();
+ InProgressSequences.finishInProgressSequences(self);
+ }
+
+ static void abortMove(String nodeId)
+ {
+ abortHelper(nodeId, MultiStepOperation.Kind.MOVE, MOVE_FAILED);
+ }
+
+ /**
+ *
+ * @param nodeId node id to abort the MSO for, null for local node
+ * @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)
+ {
+ if (ClusterMetadataService.instance().isMigrating() || ClusterMetadataService.state() == ClusterMetadataService.State.GOSSIP)
+ throw new IllegalStateException(String.format("This cluster is migrating to cluster metadata, can't abort %s until that is done.", kind));
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+ NodeId toAbort = nodeId == null ? metadata.myNodeId() : NodeId.fromString(nodeId);
+ MultiStepOperation<?> sequence = metadata.inProgressSequences.get(toAbort);
+ if (sequence == null || sequence.kind() != kind)
+ {
+ 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()))
+ {
+ if (ssMode != null && StorageService.instance.operationMode() != ssMode)
+ {
+ String msg = String.format("Can't abort a %s operation unless it has failed", kind);
+ logger.info(msg);
+ throw new IllegalStateException(msg);
+ }
+ StorageService.instance.clearTransientMode();
+ }
+ else if (Gossiper.instance.isAlive(metadata.directory.endpoint(toAbort)))
+ {
+ String msg = String.format("Can't abort a %s operation for a node %s (%s) that is UP - run abortdecommission on that instance",
+ kind, toAbort, metadata.directory.endpoint(toAbort));
+ logger.info(msg);
+ throw new IllegalStateException(msg);
+ }
+ ClusterMetadataService.instance().commit(new CancelInProgressSequence(toAbort));
+ }
+}
diff --cc src/java/org/apache/cassandra/tcm/sequences/UnbootstrapStreams.java
index 4a38530753,0000000000..72198aa05b
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapStreams.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/UnbootstrapStreams.java
@@@ -1,236 -1,0 +1,235 @@@
+/*
+ * 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.sequences;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.batchlog.BatchlogManager;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.hints.HintsService;
+import org.apache.cassandra.locator.EndpointsByReplica;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.RangesAtEndpoint;
+import org.apache.cassandra.locator.RangesByEndpoint;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.locator.SystemStrategy;
+import org.apache.cassandra.schema.Keyspaces;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.streaming.StreamOperation;
+import org.apache.cassandra.streaming.StreamPlan;
+import org.apache.cassandra.streaming.StreamState;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.ownership.MovementMap;
+import org.apache.cassandra.tcm.ownership.PlacementDeltas;
+import org.apache.cassandra.utils.concurrent.Future;
+import org.apache.cassandra.utils.concurrent.ImmediateFuture;
+
+public class UnbootstrapStreams implements LeaveStreams
+{
+ private static final Logger logger = LoggerFactory.getLogger(UnbootstrapStreams.class);
+ private final AtomicBoolean started = new AtomicBoolean();
+
+ public Kind kind()
+ {
+ return Kind.UNBOOTSTRAP;
+ }
+
+ @Override
+ public void execute(NodeId leaving, PlacementDeltas startLeave, PlacementDeltas midLeave, PlacementDeltas finishLeave) throws ExecutionException, InterruptedException
+ {
+ MovementMap movements = movementMap(ClusterMetadata.current().directory.endpoint(leaving),
+ startLeave,
+ finishLeave);
+ movements.forEach((params, eps) -> logger.info("Unbootstrap movements: {}: {}", params, eps));
+ started.set(true);
+ try
+ {
+ unbootstrap(Schema.instance.getNonLocalStrategyKeyspaces(), movements);
+ }
+ catch (ExecutionException e)
+ {
+ logger.error("Error while decommissioning node", e);
+ throw e;
+ }
+ }
+
+ private static MovementMap movementMap(InetAddressAndPort leaving, PlacementDeltas startDelta, PlacementDeltas finishDelta)
+ {
+ MovementMap.Builder allMovements = MovementMap.builder();
+ // map of src->dest movements, keyed by replication settings. During unbootstrap, this will be used to construct
+ // a stream plan for each keyspace, based on their replication params.
+ finishDelta.forEach((params, delta) -> {
+ // no streaming for LocalStrategy and friends
+ if (SystemStrategy.class.isAssignableFrom(params.klass))
+ return;
+
+ // first identify ranges to be migrated off the leaving node
+ Map<Range<Token>, Replica> oldReplicas = delta.writes.removals.get(leaving).byRange();
+
+ // next go through the additions to the write groups that will be applied during the
+ // first step of the plan. These represent the ranges moving to new replicas so in
+ // order to construct a streaming plan we can match these up with the corresponding
+ // removals to produce a src->dest mapping.
+ EndpointsByReplica.Builder movements = new EndpointsByReplica.Builder();
+ RangesByEndpoint startWriteAdditions = startDelta.get(params).writes.additions;
+ RangesByEndpoint startWriteRemovals = startDelta.get(params).writes.removals;
+ startWriteAdditions.flattenValues()
+ .forEach(newReplica -> {
+ if (startWriteRemovals.get(newReplica.endpoint()).contains(newReplica.range(), false))
+ logger.debug("Streaming transient -> full conversion to {} from {}", newReplica, oldReplicas.get(newReplica.range()));
+ movements.put(oldReplicas.get(newReplica.range()), newReplica);
+ });
+ allMovements.put(params, movements.build());
+ });
+ return allMovements.build();
+ }
+
+ private static void unbootstrap(Keyspaces keyspaces, MovementMap movements) throws ExecutionException, InterruptedException
+ {
+ Supplier<Future<StreamState>> startStreaming = prepareUnbootstrapStreaming(keyspaces, movements);
+
+ StorageService.instance.repairPaxosForTopologyChange("decommission");
+ logger.info("replaying batch log and streaming data to other nodes");
+ // Start with BatchLog replay, which may create hints but no writes since this is no longer a valid endpoint.
+ Future<?> batchlogReplay = BatchlogManager.instance.startBatchlogReplay();
+ Future<StreamState> streamSuccess = startStreaming.get();
+
+ // Wait for batch log to complete before streaming hints.
+ logger.debug("waiting for batch log processing.");
+ batchlogReplay.get();
+
+ Future<?> hintsSuccess = ImmediateFuture.success(null);
+
+ if (DatabaseDescriptor.getTransferHintsOnDecommission())
+ {
+ logger.info("streaming hints to other nodes");
+ hintsSuccess = StorageService.instance.streamHints();
+ }
+ else
+ {
+ logger.info("pausing dispatch and deleting hints");
+ DatabaseDescriptor.setHintedHandoffEnabled(false);
+ HintsService.instance.pauseDispatch();
+ HintsService.instance.deleteAllHints();
+ }
+
+ // wait for the transfer runnables to signal the latch.
+ logger.debug("waiting for stream acks.");
+ streamSuccess.get();
+ hintsSuccess.get();
+
+ logger.debug("stream acks all received.");
+ }
+
+ private static Supplier<Future<StreamState>> prepareUnbootstrapStreaming(Keyspaces keyspaces,
+ MovementMap movements)
+ {
+ // PrepareLeave transformation gives us a map of range movements for unbootstrap, keyed on replication settings.
+ // The movements themselves are maps of leavingReplica -> newReplica(s). Here we just "inflate" the outer
+ // map to a set of movements per-keyspace, duplicating where keyspaces share the same replication params
+ Map<String, EndpointsByReplica> byKeyspace =
+ keyspaces.stream()
+ .collect(Collectors.toMap(k -> k.name,
+ k -> movements.get(k.params.replication)));
+
+ return () -> streamRanges(byKeyspace);
+ }
+
+ /**
+ * Send data to the endpoints that will be responsible for it in the future
+ *
+ * @param rangesToStreamByKeyspace keyspaces and data ranges with endpoints included for each
+ * @return async Future for whether stream was success
+ */
+ private static Future<StreamState> streamRanges(Map<String, EndpointsByReplica> rangesToStreamByKeyspace)
+ {
+ // First, we build a list of ranges to stream to each host, per table
+ Map<String, RangesByEndpoint> sessionsToStreamByKeyspace = new HashMap<>();
+
+ for (Map.Entry<String, EndpointsByReplica> entry : rangesToStreamByKeyspace.entrySet())
+ {
+ String keyspace = entry.getKey();
+ EndpointsByReplica rangesWithEndpoints = entry.getValue();
+
+ if (rangesWithEndpoints.isEmpty())
+ continue;
+
- //Description is always Unbootstrap? Is that right?
- Map<InetAddressAndPort, Set<Range<Token>>> transferredRangePerKeyspace = SystemKeyspace.getTransferredRanges("Unbootstrap",
++ Map<InetAddressAndPort, Set<Range<Token>>> transferredRangePerKeyspace = SystemKeyspace.getTransferredRanges(StreamOperation.DECOMMISSION,
+ keyspace,
+ ClusterMetadata.current().tokenMap.partitioner());
+ RangesByEndpoint.Builder replicasPerEndpoint = new RangesByEndpoint.Builder();
+ for (Map.Entry<Replica, Replica> endPointEntry : rangesWithEndpoints.flattenEntries())
+ {
+ Replica local = endPointEntry.getKey();
+ Replica remote = endPointEntry.getValue();
+ Set<Range<Token>> transferredRanges = transferredRangePerKeyspace.get(remote.endpoint());
+ if (transferredRanges != null && transferredRanges.contains(local.range()))
+ {
- logger.debug("Skipping transferred range {} of keyspace {}, endpoint {}", local, keyspace, remote);
++ logger.info("Skipping transferred range {} of keyspace {}, endpoint {}", local, keyspace, remote);
+ continue;
+ }
+
+ replicasPerEndpoint.put(remote.endpoint(), remote.decorateSubrange(local.range()));
+ }
+
+ sessionsToStreamByKeyspace.put(keyspace, replicasPerEndpoint.build());
+ }
+
+ StreamPlan streamPlan = new StreamPlan(StreamOperation.DECOMMISSION);
+
+ // Vinculate StreamStateStore to current StreamPlan to update transferred ranges per StreamSession
+ streamPlan.listeners(StorageService.instance.streamStateStore());
+
+ for (Map.Entry<String, RangesByEndpoint> entry : sessionsToStreamByKeyspace.entrySet())
+ {
+ String keyspaceName = entry.getKey();
+ RangesByEndpoint replicasPerEndpoint = entry.getValue();
+
+ for (Map.Entry<InetAddressAndPort, RangesAtEndpoint> rangesEntry : replicasPerEndpoint.asMap().entrySet())
+ {
+ RangesAtEndpoint replicas = rangesEntry.getValue();
+ InetAddressAndPort newEndpoint = rangesEntry.getKey();
+
+ // TODO each call to transferRanges re-flushes, this is potentially a lot of waste
+ streamPlan.transferRanges(newEndpoint, keyspaceName, replicas);
+ }
+ }
+ return streamPlan.execute();
+ }
+
+ // todo: add more details
+ public String status()
+ {
+ return "streams" + (started.get() ? "" : " not") + " started";
+ }
+}
diff --cc test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
index 8012399afd,f70c10574b..d55d550b71
--- a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java
@@@ -256,25 -328,19 +256,30 @@@ public class BootstrapTest extends Test
}
public static void populate(ICluster cluster, int from, int to, int coord, int rf, ConsistencyLevel cl)
+ {
+ populate(cluster, from, to, coord, rf, cl, "pk int, ck int, v int");
+ }
+
+ public static void populate(ICluster cluster, int from, int to, int coord, int rf, ConsistencyLevel cl, String columnDefinitions)
{
cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + rf + "};");
- cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+ cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl (" + columnDefinitions + ", PRIMARY KEY (pk, ck))");
+ populateExistingTable(cluster, from, to, coord, cl);
+ for (int i = from; i < to; i++)
+ {
+ cluster.coordinator(coord).executeWithRetries("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)",
+ cl,
+ i, i, i);
+ }
+ }
+
+ public static void populateExistingTable(ICluster cluster, int from, int to, int coord, ConsistencyLevel cl)
+ {
for (int i = from; i < to; i++)
{
- cluster.coordinator(coord).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)",
- cl,
- i, i, i);
+ cluster.coordinator(coord).executeWithRetries("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)",
+ cl,
+ i, i, i);
}
}
diff --cc test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
index 69b8e31430,0000000000..01d4b94fb5
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java
@@@ -1,316 -1,0 +1,410 @@@
+/*
+ * 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.io.IOException;
++import java.nio.ByteBuffer;
++import java.util.Arrays;
+import java.util.HashSet;
++import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
++import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.QueryProcessor;
++import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.Constants;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.streaming.StreamSession;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.transformations.PrepareLeave;
+import org.apache.cassandra.tcm.transformations.Startup;
+import org.apache.cassandra.utils.CassandraVersion;
+import org.apache.cassandra.utils.FBUtilities;
++import org.apache.cassandra.utils.concurrent.Future;
++import org.apache.cassandra.utils.concurrent.ImmediateFuture;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
++import static org.apache.cassandra.db.SystemKeyspace.TRANSFERRED_RANGES_V2;
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.pauseBeforeCommit;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.unpauseCommits;
+import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack;
+import static org.apache.cassandra.distributed.shared.NetworkTopology.networkTopology;
+import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class DecommissionTest extends TestBaseImpl
+{
+ @Test
+ public void testResumableDecom() throws IOException
+ {
+ try (Cluster cluster = builder().withNodes(3)
+ .withConfig(config -> config.with(NETWORK, GOSSIP))
+ .withInstanceInitializer(BB::install)
+ .start())
+ {
+ populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM);
+ cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure();
+ cluster.get(2).nodetoolResult("decommission", "--force").asserts().success();
+ }
+ }
+
+ @Test
+ public void testOperationModeOnDecomResume() throws Exception
+ {
+ // Node 2's first decomission attempt fails mid-stream (injected by BB), leaving it in
+ // DECOMISSION_FAILED. On resume, operationMode() must transition back to LEAVING before
+ // MID_LEAVE is committed. We pause the CMS just before that commit to assert the mode
+ // in the window where streaming has finished but the epoch has not yet advanced.
+ try (Cluster cluster = builder().withNodes(3)
+ .withConfig(config -> config.with(NETWORK, GOSSIP))
+ .withInstanceInitializer(BB::install)
+ .start())
+ {
+ populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM);
+
+ IInvokableInstance cmsNode = cluster.get(1);
+ IInvokableInstance leavingNode = cluster.get(2);
+
- // --force required: cie_internal keyspace has RF=3, decommission would fail replication check otherwise
++ // --force required: system_distributed keyspace has RF=3, decommission would fail replication check otherwise
+ leavingNode.nodetoolResult("decommission", "--force").asserts().failure();
+ leavingNode.runOnInstance(() -> assertEquals(StorageService.Mode.DECOMMISSION_FAILED, StorageService.instance.operationMode()));
+
+ Callable<Epoch> midLeavePaused = pauseBeforeCommit(cmsNode, e -> e instanceof PrepareLeave.MidLeave);
+
+ Thread resumeThread = new Thread(() -> leavingNode.nodetoolResult("decommission").asserts().success());
+ resumeThread.start();
+ midLeavePaused.call();
+
+ leavingNode.runOnInstance(() ->
+ assertEquals("operationMode during resumed decommission streaming should be LEAVING, not DECOMMISSION_FAILED",
+ StorageService.Mode.LEAVING,
+ StorageService.instance.operationMode()));
+
+ unpauseCommits(cmsNode);
+ resumeThread.join(TimeUnit.MINUTES.toMillis(2));
+ }
+ }
+
+ @Test
+ public void testAddressReuseAfterDecommission() throws IOException, ExecutionException, InterruptedException
+ {
+ // Initially, all nodes should be in dc1/rack1. Node 3 will be decommissioned and a new node added re-using
+ // node 3's address. When the new node registers, it should be in dc2/rack2.
+ // For now, this requires the accord service to disabled. See CASSANDRA-21026
+ try (Cluster cluster = builder().withNodes(3)
+ .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4))
+ .withConfig(config -> config.with(NETWORK, GOSSIP)
+ .set("accord.enabled", false))
+ .withNodeIdTopology(networkTopology(3, (id) -> dcAndRack("dc1", "rack1")))
+ .start())
+ {
+ assertEquals("dc1/rack1", cluster.get(1).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc1/rack1", cluster.get(2).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc1/rack1", cluster.get(3).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+
+ IInvokableInstance toRemove = cluster.get(3);
+ toRemove.nodetoolResult("decommission", "--force").asserts().success();
+ toRemove.shutdown().get();
+ ClusterUtils.getDirectories(toRemove).forEach(File::tryDeleteRecursive);
+ cluster.unsafeRemoveNode(toRemove);
+
+ // Now add a new node, using the same address as the one we just removed. This new node should register
+ // itself in dc2/rack2 and not inherit the location of its predecessor.
+ // Note: because we have removed the original node3 from the cluster completely, which is necessary because
+ // the cluster will complain about an id clash otherwise, this new node will also be "node3". However, it is
+ // completely distinct from the original one.
+ cluster.unsafeUpdateNodeIdTopology(toRemove.config().num(), dcAndRack("dc2", "rack2"));
+ IInstanceConfig config = cluster.newInstanceConfig()
+ .set("auto_bootstrap", true)
+ .set(Constants.KEY_DTEST_FULL_STARTUP, true);
+ IInvokableInstance newInstance = cluster.bootstrap(config);
+ newInstance.startup();
+
+ assertEquals("dc1/rack1", cluster.get(1).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc1/rack1", cluster.get(2).callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ assertEquals("dc2/rack2", newInstance.callOnInstance(() -> DatabaseDescriptor.getLocator().local().toString()));
+ }
+ }
+
++ @Test
++ public void testAbortingDecommissionRestreams() throws Exception
++ {
++ // https://issues.apache.org/jira/browse/CASSANDRA-16290
++ // We demonstrate here that decommissioning and then aborting decommission is unsafe
++ // if we've persisted transferred ranges and then skip them for something which was delivered after we aborted the decommission but before we resumed
++ try (Cluster cluster = builder().withNodes(4)
++ .withConfig(config -> config.with(NETWORK, GOSSIP)
++ // disable hints to simplify test
++ .set("hinted_handoff_enabled", false)
++ )
++ .withInstanceInitializer(BB::streamHintsInstall)
++ .start())
++ {
++ // We need blob columns here so later we can do Murmur3Partitioner.LongToken.keyForToken(token);
++ populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM, "pk blob, ck blob, v blob");
++
++ IInvokableInstance leavingNode = cluster.get(2);
++
++ leavingNode.nodetoolResult("decommission").asserts().failure();
++ leavingNode.runOnInstance(() -> assertEquals(StorageService.Mode.DECOMMISSION_FAILED, StorageService.instance.operationMode()));
++
++ // abort the decommission
++ leavingNode.nodetoolResult("abortdecommission").asserts().success();
++
++ // Stop the non leaving nodes so we can write at ONE and fail to stream that datum
++ ClusterUtils.stopUnchecked(cluster.get(1));
++ ClusterUtils.stopUnchecked(cluster.get(3));
++ ClusterUtils.stopUnchecked(cluster.get(4));
++
++ List<Murmur3Partitioner.LongToken> tokens = ClusterUtils.getLocalTokens(leavingNode).stream().map(t -> new Murmur3Partitioner.LongToken(Long.parseLong(t))).collect(Collectors.toList());
++ for (Murmur3Partitioner.LongToken token : tokens)
++ {
++ ByteBuffer key = Murmur3Partitioner.LongToken.keyForToken(token);
++ leavingNode.coordinator().execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)", ConsistencyLevel.ONE, key, key, key);
++ }
++
++ ClusterUtils.start(cluster.get(1), props -> {});
++ ClusterUtils.start(cluster.get(3), props -> {});
++ ClusterUtils.start(cluster.get(4), props -> {});
++
++ ClusterUtils.awaitRingHealthy(leavingNode);
++ ClusterUtils.waitForCMSToQuiesce(cluster, cluster.get(1));
++
++ Object[][] ranges = leavingNode.executeInternal("SELECT keyspace_name from system." + TRANSFERRED_RANGES_V2);
++
++ assertTrue("transferred ranges missing entirely", ranges.length > 0);
++ assertTrue("transferred ranges present for keyspace", Arrays.stream(ranges).anyMatch(x -> x[0].equals(KEYSPACE)));
++
++ // Resume decomm
++ leavingNode.nodetoolResult("decommission").asserts().success();
++
++ // Try and read data we wrote at ONE at ALL
++ for (Murmur3Partitioner.LongToken token : tokens)
++ {
++ ByteBuffer key = Murmur3Partitioner.LongToken.keyForToken(token);
++ Object[][] resp = cluster.get(1).coordinator().execute("SELECT pk from " + KEYSPACE + ".tbl where pk=?", ConsistencyLevel.ALL, key);
++ assertTrue("We should get a response for this key we wrote it at ONE", resp.length > 0);
++ assertEquals(key, resp[0][0]);
++ }
++ }
++ }
++
+ public static class BB
+ {
++
+ static void install(ClassLoader cl, int nodeNumber)
+ {
+ if (nodeNumber != 2)
+ return;
+ new ByteBuddy().rebase(StreamSession.class)
+ .method(named("startStreamingFiles"))
+ .intercept(MethodDelegation.to(BB.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ }
++
++ static void streamHintsInstall(ClassLoader cl, int nodeNumber)
++ {
++ if (nodeNumber != 2)
++ return;
++ new ByteBuddy().rebase(StorageService.class)
++ .method(named("streamHints"))
++ .intercept(MethodDelegation.to(BB.class))
++ .make()
++ .load(cl, ClassLoadingStrategy.Default.INJECTION);
++ }
++
+ static AtomicBoolean first = new AtomicBoolean();
+
++ public static Future<?> streamHints(@SuperCall Callable<Future<?>> zuper) throws Exception
++ {
++ if (!first.get())
++ {
++ first.set(true);
++ return ImmediateFuture.failure(new IOException("failing hints so that decomm fails at last moment possible" ));
++ }
++ return zuper.call();
++ }
++
+ public static void startStreamingFiles(@Nullable StreamSession.PrepareDirection prepareDirection, @SuperCall Callable<Void> zuper) throws Exception
+ {
+ if (!first.get())
+ {
+ first.set(true);
+ throw new RuntimeException("Triggering streaming error");
+ }
+ zuper.call();
+ }
+ }
+
+ @Test
+ public void testAbortDecom() throws IOException
+ {
+ try (Cluster cluster = builder().withNodes(3)
+ .withConfig(config -> config.with(NETWORK, GOSSIP))
+ .withInstanceInitializer(BB::install)
+ .start())
+ {
+ populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM);
+ cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure();
+ cluster.get(2).nodetoolResult("abortdecommission").asserts().success();
+ cluster.get(2).runOnInstance(() -> {
+ assertEquals(StorageService.Mode.NORMAL, StorageService.instance.operationMode());
+ assertTrue(ClusterMetadata.current().inProgressSequences.isEmpty());
+ });
+ cluster.get(2).nodetoolResult("decommission", "--force").asserts().success();
+ }
+ }
+
+ @Test
+ public void testAbortDecomRemote() throws IOException, ExecutionException, InterruptedException
+ {
+ try (Cluster cluster = builder().withNodes(3)
+ .withConfig(config -> config.with(NETWORK, GOSSIP))
+ .withInstanceInitializer(BB::install)
+ .start())
+ {
+ populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM);
+ int nodeId = cluster.get(2).callOnInstance(() -> {
+ return ClusterMetadata.current().myNodeId().id();
+ });
+ cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure();
+ cluster.get(2).shutdown().get();
+ cluster.get(3).runOnInstance(() -> {
+ while (Gossiper.instance.isAlive(ClusterMetadata.current().directory.endpoint(new NodeId(nodeId))))
+ Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
+ });
+ cluster.get(3).nodetoolResult("abortdecommission", "--node", String.valueOf(nodeId)).asserts().success();
+ cluster.get(2).startup();
+ cluster.get(2).runOnInstance(() -> {
+ assertEquals(StorageService.Mode.NORMAL, StorageService.instance.operationMode());
+ assertTrue(ClusterMetadata.current().inProgressSequences.isEmpty());
+ });
+ cluster.get(2).runOnInstance(() -> {
+ BB.first.set(true);
+ });
+ cluster.get(2).nodetoolResult("decommission", "--force").asserts().success();
+ }
+ }
+
+ @Test
+ public void testDecomDirectoryMinMaxVersions() throws IOException {
+ try (Cluster cluster = builder()
+ .withConfig(cfg -> cfg.with(GOSSIP))
+ .withNodes(3)
+ .start())
+ {
+ cluster.get(3).nodetoolResult("decommission", "--force").asserts().success();
+
+ cluster.get(1).runOnInstance(() -> {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ ClusterMetadataService.instance().commit(new Startup(metadata.myNodeId(),
+ metadata.directory.getNodeAddresses(metadata.myNodeId()),
+ new NodeVersion(new CassandraVersion("6.0.0"),
+ NodeVersion.CURRENT_METADATA_VERSION)));
+ });
+
+ cluster.get(2).runOnInstance(() -> {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ ClusterMetadataService.instance().commit(new Startup(metadata.myNodeId(),
+ metadata.directory.getNodeAddresses(metadata.myNodeId()),
+ new NodeVersion(new CassandraVersion("5.0.0"),
+ NodeVersion.CURRENT_METADATA_VERSION)));
+ });
+
+ for (int i = 1; i <= 2; i++)
+ {
+ cluster.get(i).runOnInstance(() -> {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ assertEquals(new CassandraVersion("5.0.0"), metadata.directory.clusterMinVersion.cassandraVersion);
+ assertEquals(new CassandraVersion("6.0.0"), metadata.directory.clusterMaxVersion.cassandraVersion);
+ assertTrue(metadata.directory.versions.containsValue(NodeVersion.CURRENT));
+ });
+ }
+ }
+ }
+
+ @Test
+ public void testPeersPostDecom() throws IOException
+ {
+ try (Cluster cluster = builder().withNodes(4)
+ .withConfig(config -> config.with(NETWORK, GOSSIP))
+ .start())
+ {
+ populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM);
+ cluster.get(3).nodetoolResult("decommission", "--force").asserts().success();
+
+ int[] remainingNodes = {1, 2, 4};
+ Set<String> expectedPeers = new HashSet<>();
+ for (int i : remainingNodes)
+ expectedPeers.add(cluster.get(i).config().broadcastAddress().getAddress().toString());
+
+ // Decommission should remove from both the peers & peers_v2 system tables
+ for (int i : remainingNodes)
+ {
+ cluster.get(i).runOnInstance(() -> {
+ for (String table : new String[] {"peers", "peers_v2"})
+ {
+ Set<String> values = new HashSet<>();
+ QueryProcessor.executeInternal(String.format("SELECT peer from system.%s;", table))
+ .forEach(r -> values.add(r.getInetAddress("peer").toString()));
+ assertEquals(2, values.size());
+ for (String e : expectedPeers)
+ if (!e.equals(FBUtilities.getJustBroadcastAddress().toString()))
+ assertTrue(values.contains(e));
+ }
+ });
+ }
+ }
+ }
+
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org