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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
