natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Don’t finish ongoing decommission and move operations during startup 

Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-20040
This commit is contained in:
Marcus Eriksson 2024-10-22 11:27:44 +02:00
parent 73d8ee1a71
commit c679b47303
15 changed files with 390 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Don’t finish ongoing decommission and move operations during startup (CASSANDRA-20040)
* Nodetool reconfigure cms has correct return code when streaming fails (CASSANDRA-19972)
* Reintroduce RestrictionSet#iterator() optimization around multi-column restrictions (CASSANDRA-20034)
* Explicitly localize strings to Locale.US for internal implementation (CASSANDRA-19953)

38
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -253,6 +253,8 @@ import static org.apache.cassandra.service.ActiveRepairService.repairCommandExec
import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSIONED;
import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSION_FAILED;
import static org.apache.cassandra.service.StorageService.Mode.JOINING_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.tcm.membership.NodeState.BOOTSTRAPPING;
import static org.apache.cassandra.tcm.membership.NodeState.BOOT_REPLACING;
@ -454,8 +456,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
/* the probability for tracing any particular request, 0 disables tracing and 1 enables for all */
private double traceProbability = 0.0;
public enum Mode { STARTING, NORMAL, JOINING, JOINING_FAILED, LEAVING, DECOMMISSIONED, DECOMMISSION_FAILED, MOVING, DRAINING, DRAINED }
private volatile Mode operationMode = Mode.STARTING;
public enum Mode { STARTING, NORMAL, JOINING, JOINING_FAILED, LEAVING, DECOMMISSIONED, DECOMMISSION_FAILED, MOVING, MOVE_FAILED, DRAINING, DRAINED }
/* Can currently hold DECOMMISSIONED, DECOMMISSION_FAILED, DRAINING, DRAINED for legacy compatibility. */
private volatile Optional<Mode> transientMode = Optional.empty();
@ -763,7 +764,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
});
if (SystemKeyspace.wasDecommissioned())
throw new ConfigurationException("This node was decommissioned and will not rejoin the ring unless cassandra.override_decommission=true has been set, or all existing data is removed and the node is bootstrapped again");
{
if (CassandraRelevantProperties.OVERRIDE_DECOMMISSION.getBoolean())
{
logger.warn("This node was decommissioned, but overriding by operator request.");
}
else
{
throw new ConfigurationException("This node was decommissioned and will not rejoin the ring unless cassandra.override_decommission=true has been set, or all existing data is removed and the node is bootstrapped again");
}
}
if (DatabaseDescriptor.getReplaceTokens().size() > 0 || DatabaseDescriptor.getReplaceNode() != null)
throw new RuntimeException("Replace method removed; use cassandra.replace_address instead");
@ -3669,6 +3679,18 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
SingleNodeSequences.move(getTokenFactory().fromString(newToken));
}
@Override
public void resumeMove()
{
SingleNodeSequences.resumeMove();
}
@Override
public void abortMove()
{
SingleNodeSequences.abortMove();
}
public String getRemovalStatus()
{
return getRemovalStatus(false);
@ -3776,6 +3798,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
transientMode = Optional.of(JOINING_FAILED);
}
public void markMoveFailed()
{
logger.info(MOVE_FAILED.toString());
transientMode = Optional.of(MOVE_FAILED);
}
/*
- Use system_views.local to get information about the node (todo: we might still need a jmx endpoint for that since you can't run cql queries on drained etc nodes)
*/
@ -3845,7 +3873,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public boolean isDecommissioned()
{
return operationMode == DECOMMISSIONED;
return operationMode() == DECOMMISSIONED;
}
public boolean isDecommissionFailed()
@ -3855,7 +3883,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public boolean isDecommissioning()
{
return operationMode == Mode.LEAVING || operationMode == DECOMMISSION_FAILED;
return operationMode() == LEAVING;
}
public boolean isBootstrapFailed()

2
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -530,6 +530,8 @@ public interface StorageServiceMBean extends NotificationEmitter
* This node will unload its data onto its neighbors, and bootstrap to the new token.
*/
public void move(String newToken) throws IOException;
public void resumeMove();
public void abortMove();
/**
* removeToken removes token (and all data associated with

5
src/java/org/apache/cassandra/tcm/MultiStepOperation.java
View File

@ -98,6 +98,11 @@ public abstract class MultiStepOperation<CONTEXT>
this.latestModification = latestModification;
}
public boolean finishDuringStartup()
{
return true;
}
/**
* Unique identifier for the type of operation, e.g. JOIN, LEAVE, MOVE
* @return the specific kind of this operation

13
src/java/org/apache/cassandra/tcm/Startup.java
View File

@ -396,7 +396,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
NodeId self = metadata.myNodeId();
// finish in-progress sequences first
InProgressSequences.finishInProgressSequences(self);
InProgressSequences.finishInProgressSequences(self, true);
metadata = ClusterMetadata.current();
switch (metadata.directory.peerState(self))
@ -407,8 +407,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
ReconfigureCMS.maybeReconfigureCMS(metadata, DatabaseDescriptor.getReplaceAddress());
ClusterMetadataService.instance().commit(initialTransformation.get());
InProgressSequences.finishInProgressSequences(self);
InProgressSequences.finishInProgressSequences(self, true); // potentially finish the MSO committed above
metadata = ClusterMetadata.current();
if (metadata.directory.peerState(self) == JOINED)
@ -437,6 +436,14 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
"Can't proceed from the state " + metadata.directory.peerState(self));
}
break;
case LEAVING:
logger.info("Node is currently being decommissioned, resume with `nodetool decommission`");
StorageService.instance.markDecommissionFailed();
break;
case MOVING:
logger.info("Node is currently moving, resume with nodetool move --resume or abort with nodetool move --abort");
StorageService.instance.markMoveFailed();
break;
default:
throw new IllegalStateException("Can't proceed from the state " + metadata.directory.peerState(self));
}

7
src/java/org/apache/cassandra/tcm/sequences/InProgressSequences.java
View File

@ -60,6 +60,11 @@ public class InProgressSequences implements MetadataValue<InProgressSequences>,
}
public static void finishInProgressSequences(MultiStepOperation.SequenceKey sequenceKey)
{
finishInProgressSequences(sequenceKey, false);
}
public static void finishInProgressSequences(MultiStepOperation.SequenceKey sequenceKey, boolean onlyStartupSafeSequences)
{
ClusterMetadata metadata = ClusterMetadata.current();
while (true)
@ -67,6 +72,8 @@ public class InProgressSequences implements MetadataValue<InProgressSequences>,
MultiStepOperation<?> sequence = metadata.inProgressSequences.get(sequenceKey);
if (sequence == null)
break;
if (onlyStartupSafeSequences && !sequence.finishDuringStartup())
break;
if (isLeave(sequence))
StorageService.instance.maybeInitializeServices();
if (resume(sequence))

14
src/java/org/apache/cassandra/tcm/sequences/Move.java
View File

@ -151,6 +151,12 @@ public class Move extends MultiStepOperation<Epoch>
this.streamData = current.streamData;
}
@Override
public boolean finishDuringStartup()
{
return false;
}
@Override
public Kind kind()
{
@ -199,7 +205,7 @@ public class Move extends MultiStepOperation<Epoch>
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
return continuable() ;
return continuable();
}
break;
case MID_MOVE:
@ -251,8 +257,14 @@ public class Move extends MultiStepOperation<Epoch>
}
catch (ExecutionException e)
{
StorageService.instance.markMoveFailed();
throw new RuntimeException("Unable to move", e);
}
catch (Exception e)
{
StorageService.instance.markMoveFailed();
throw e;
}
try
{

57
src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java
View File

@ -33,6 +33,7 @@ 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;
@ -75,13 +76,17 @@ public interface SingleNodeSequences
if (inProgress == null)
{
logger.info("starting decom with {} {}", metadata.epoch, self);
logger.info("starting decommission with {} {}", metadata.epoch, self);
ClusterMetadataService.instance().commit(new PrepareLeave(self,
force,
ClusterMetadataService.instance().placementProvider(),
LeaveStreams.Kind.UNBOOTSTRAP));
}
else if (!InProgressSequences.isLeave(inProgress))
else if (InProgressSequences.isLeave(inProgress))
{
logger.info("Resuming decommission @ {} (current epoch = {}): {}", inProgress.latestModification, metadata.epoch, inProgress.status());
}
else
{
throw new IllegalArgumentException("Can not decommission a node that has an in-progress sequence");
}
@ -165,4 +170,52 @@ public interface SingleNodeSequences
logger.debug("Successfully moved to new token {}", StorageService.instance.getLocalTokens().iterator().next());
}
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);
throw new IllegalStateException(msg);
}
if (StorageService.instance.operationMode() != StorageService.Mode.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()
{
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 abort";
logger.info(msg);
throw new IllegalStateException(msg);
}
if (StorageService.instance.operationMode() != StorageService.Mode.MOVE_FAILED)
{
String msg = "Can't abort a move operation unless it has failed";
logger.info(msg);
throw new IllegalStateException(msg);
}
StorageService.instance.clearTransientMode();
ClusterMetadataService.instance().commit(new CancelInProgressSequence(self));
}
}

6
src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java
View File

@ -292,6 +292,12 @@ public class UnbootstrapAndLeave extends MultiStepOperation<Epoch>
}
}
@Override
public boolean finishDuringStartup()
{
return false;
}
@Override
public String toString()
{

10
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -1010,6 +1010,16 @@ public class NodeProbe implements AutoCloseable
ssProxy.move(newToken);
}
public void resumeMove()
{
ssProxy.resumeMove();
}
public void abortMove()
{
ssProxy.abortMove();
}
public void removeNode(String token)
{
removeNode(token, false);

29
src/java/org/apache/cassandra/tools/nodetool/Move.java
View File

@ -23,21 +23,46 @@ import io.airlift.airline.Command;
import java.io.IOException;
import io.airlift.airline.Option;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
@Command(name = "move", description = "Move node on the token ring to a new token")
public class Move extends NodeToolCmd
{
@Arguments(usage = "<new token>", description = "The new token.", required = true)
@Arguments(usage = "<new token>", description = "The new token.")
private String newToken = EMPTY;
@Option(title = "Resume an ongoing move operation", name = "--resume")
private boolean resume;
@Option(title = "Abort an ongoing move operation", name = "--abort")
private boolean abort;
@Override
public void execute(NodeProbe probe)
{
try
{
probe.move(newToken);
if (!newToken.isEmpty())
{
if (resume || abort)
throw new IllegalArgumentException("Can't give both a token and --resume/--abort");
probe.move(newToken);
}
else
{
if (abort && resume)
throw new IllegalArgumentException("Can't both resume and abort");
if (resume)
probe.resumeMove();
else if (abort)
probe.abortMove();
else
throw new IllegalArgumentException("Need to give either a token for a new move operation, or --resume/--abort for an existing one");
}
} catch (IOException e)
{
throw new RuntimeException("Error during moving node", e);

1
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -636,6 +636,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
}
catch (Throwable t)
{
startedAt.set(0);
if (t instanceof RuntimeException)
throw (RuntimeException) t;
throw new RuntimeException(t);

163
test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.distributed.test;
import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
@ -27,9 +28,13 @@ import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.action.GossipHelper;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.PlacementDeltas;
import org.apache.cassandra.tcm.sequences.UnbootstrapStreams;
@ -41,7 +46,6 @@ 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.stopUnchecked;
import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSION_FAILED;
import static org.apache.cassandra.service.StorageService.Mode.NORMAL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@ -59,62 +63,31 @@ public class DecommissionTest extends TestBaseImpl
.start()))
{
IInvokableInstance instance = cluster.get(2);
assertBootstrapState(instance, COMPLETED);
instance.nodetoolResult("decommission", "--force")
.asserts()
.failure()
.stderrContains("simulated error in prepareUnbootstrapStreaming");
instance.runOnInstance(() -> {
assertFalse(StorageService.instance.isDecommissioning());
assertTrue(StorageService.instance.isDecommissionFailed());
});
assertEquals(COMPLETED.name(), StorageService.instance.getBootstrapState());
// pretend that decommissioning has failed in the middle
try
{
StorageService.instance.decommission(true);
fail("the first attempt to decommission should fail");
}
catch (Throwable t)
{
assertTrue(t.getMessage().contains("simulated error in prepareUnbootstrapStreaming"));
}
assertFalse(StorageService.instance.isDecommissioning());
assertTrue(StorageService.instance.isDecommissionFailed());
// still COMPLETED, nothing has changed
assertEquals(COMPLETED.name(), StorageService.instance.getBootstrapState());
String operationMode = StorageService.instance.getOperationMode();
assertEquals(DECOMMISSION_FAILED.name(), operationMode);
// try to decommission again, now successfully
try
{
StorageService.instance.decommission(true);
// decommission was successful, so we reset failed decommission mode
assertFalse(StorageService.instance.isDecommissionFailed());
assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
assertFalse(StorageService.instance.isDecommissioning());
}
catch (Throwable t)
{
fail("the second decommission attempt should pass but it failed on: " + t.getMessage());
}
assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
assertFalse(StorageService.instance.isDecommissionFailed());
try
{
StorageService.instance.decommission(true);
fail("Should have failed since the node is in decomissioned state");
}
catch (UnsupportedOperationException e)
{
// ignore
}
assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
// still COMPLETED, nothing has changed
assertBootstrapState(instance, COMPLETED);
assertOperationMode(instance, DECOMMISSION_FAILED);
instance.nodetoolResult("decommission", "--force").asserts().success();
instance.runOnInstance(() -> {
assertFalse(StorageService.instance.isDecommissionFailed());
assertFalse(StorageService.instance.isDecommissioning());
});
assertBootstrapState(instance, DECOMMISSIONED);
instance.nodetoolResult("decommission", "--force")
.asserts()
.success()
.stdoutContains("Node was already decommissioned");
assertBootstrapState(instance, DECOMMISSIONED);
instance.runOnInstance(() -> {
assertFalse(StorageService.instance.isDecommissionFailed());
assertFalse(StorageService.instance.isDecommissioning());
});
@ -137,46 +110,27 @@ public class DecommissionTest extends TestBaseImpl
.start()))
{
IInvokableInstance instance = cluster.get(2);
instance.runOnInstance(() -> {
assertEquals(COMPLETED.name(), StorageService.instance.getBootstrapState());
// pretend that decommissioning has failed in the middle
try
{
StorageService.instance.decommission(true);
fail("the first attempt to decommission should fail");
}
catch (Throwable t)
{
assertTrue(t.getMessage().contains("simulated error in prepareUnbootstrapStreaming"));
}
// node is in DECOMMISSION_FAILED mode
String operationMode = StorageService.instance.getOperationMode();
assertEquals(DECOMMISSION_FAILED.name(), operationMode);
});
assertBootstrapState(instance, COMPLETED);
// pretend that decommissioning has failed in the middle
instance.nodetoolResult("decommission", "--force")
.asserts()
.failure()
.stderrContains("simulated error in prepareUnbootstrapStreaming");
assertOperationMode(instance, DECOMMISSION_FAILED);
// restart the node which we failed to decommission
stopUnchecked(instance);
instance.startup();
// it is back to normal so let's decommission again
String oprationMode = instance.callOnInstance(() -> StorageService.instance.getOperationMode());
assertEquals(NORMAL.name(), oprationMode);
// it starts up as DECOMMISSION_FAILED so let's decommission again
assertOperationMode(instance, DECOMMISSION_FAILED);
instance.nodetoolResult("decommission", "--force").asserts().success();
assertBootstrapState(instance, DECOMMISSIONED);
instance.runOnInstance(() -> {
StorageService.instance.decommission(true);
assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
assertFalse(StorageService.instance.isDecommissionFailed());
assertFalse(StorageService.instance.isDecommissioning());
});
}
}
public static class BB
{
public static void install(ClassLoader classLoader, Integer num)
@ -208,4 +162,43 @@ public class DecommissionTest extends TestBaseImpl
}
}
}
@Test
public void testRestartDecommedNode() throws IOException, ExecutionException, InterruptedException
{
try (Cluster cluster = init(Cluster.build(2)
.withConfig(config -> config.with(GOSSIP)
.with(NETWORK))
.start()))
{
cluster.get(2).nodetoolResult("decommission", "--force").asserts().success();
cluster.get(2).shutdown().get();
try
{
cluster.get(2).startup();
fail();
}
catch (Exception e)
{
cluster.get(2).runOnInstance(() -> ClusterMetadataService.unsetInstance());
assertTrue(e.getMessage().contains("This node was decommissioned and will not rejoin the ring unless cassandra.override_decommission=true"));
}
GossipHelper.withProperty(CassandraRelevantProperties.OVERRIDE_DECOMMISSION, true, () -> cluster.get(2).startup());
assertBootstrapState(cluster.get(2), COMPLETED);
}
}
private static void assertBootstrapState(IInvokableInstance i, SystemKeyspace.BootstrapState expectedState)
{
String bootstrapState = expectedState.name();
i.runOnInstance(() -> assertEquals(bootstrapState, SystemKeyspace.getBootstrapState().name()));
}
private static void assertOperationMode(IInvokableInstance i, StorageService.Mode mode)
{
String operationMode = mode.name();
i.runOnInstance(() -> assertEquals(operationMode, StorageService.instance.operationMode().name()));
}
}

140
test/distributed/org/apache/cassandra/distributed/test/FailingMoveTest.java Normal file
View File

@ -0,0 +1,140 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.distributed.test;
import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamPlan;
import org.apache.cassandra.streaming.StreamResultFuture;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeId;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
public class FailingMoveTest extends TestBaseImpl
{
@Test
public void testResumeMove() throws IOException
{
try (Cluster cluster = init(Cluster.build(3)
.withoutVNodes()
.withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
.withInstanceInitializer(BB::install)
.start()))
{
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl(id int primary key);"));
for (int i=0; i<30; i++)
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (id) VALUES (?)"),
ConsistencyLevel.ALL, i);
String oldToken = getToken(cluster.get(3));
String moveToToken = "2305843009213693949";
assertNotEquals(oldToken, moveToToken);
cluster.get(3).nodetoolResult("move", moveToToken).asserts().failure();
cluster.get(3).runOnInstance(() -> {
assertEquals(StorageService.Mode.MOVE_FAILED, StorageService.instance.operationMode());
BB.shouldFail.set(false);
});
cluster.get(3).nodetoolResult("move", "--resume").asserts().success();
cluster.get(3).runOnInstance(() -> assertEquals(StorageService.Mode.NORMAL, StorageService.instance.operationMode()));
assertEquals(moveToToken, getToken(cluster.get(3)));
}
}
@Test
public void testAbortMove() throws IOException
{
try (Cluster cluster = init(Cluster.build(3)
.withoutVNodes()
.withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
.withInstanceInitializer(BB::install)
.start()))
{
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl(id int primary key);"));
for (int i=0; i<30; i++)
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (id) VALUES (?)"),
ConsistencyLevel.ALL, i);
String oldToken = getToken(cluster.get(3));
String moveToToken = "2305843009213693949";
assertNotEquals(oldToken, moveToToken);
cluster.get(3).nodetoolResult("move", moveToToken).asserts().failure();
cluster.get(3).runOnInstance(() -> {
assertEquals(StorageService.Mode.MOVE_FAILED, StorageService.instance.operationMode());
BB.shouldFail.set(false);
});
cluster.get(3).nodetoolResult("move", "--abort").asserts().success();
cluster.get(3).runOnInstance(() -> assertEquals(StorageService.Mode.NORMAL, StorageService.instance.operationMode()));
assertNotEquals(moveToToken, getToken(cluster.get(3)));
}
}
private String getToken(IInvokableInstance instance)
{
return instance.callsOnInstance(() -> {
NodeId self = ClusterMetadata.current().myNodeId();
return ClusterMetadata.current().tokenMap.tokens(self).iterator().next().toString();
}).call();
}
public static class BB
{
static AtomicBoolean shouldFail = new AtomicBoolean(true);
public static void install(ClassLoader classLoader, Integer num)
{
new ByteBuddy().rebase(StreamPlan.class)
.method(named("execute"))
.intercept(MethodDelegation.to(BB.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
}
@SuppressWarnings("unused")
public static StreamResultFuture execute(@SuperCall Callable<StreamResultFuture> zuper)
{
if (shouldFail.get())
throw new RuntimeException("failing stream");
try
{
return zuper.call();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
}
}

2
test/distributed/org/apache/cassandra/distributed/test/jmx/JMXGetterCheckTest.java
View File

@ -59,6 +59,8 @@ public class JMXGetterCheckTest extends TestBaseImpl
"org.apache.cassandra.db:type=StorageService:clearConnectionHistory", // Throws a NullPointerException
"org.apache.cassandra.db:type=StorageService:startGossiping", // causes multiple loops to fail
"org.apache.cassandra.db:type=StorageService:startNativeTransport", // causes multiple loops to fail
"org.apache.cassandra.db:type=StorageService:resumeMove", // throws since there is no move in progress
"org.apache.cassandra.db:type=StorageService:abortMove", // throws since there is no move in progress
"org.apache.cassandra.db:type=CIDRGroupsMappingManager:loadCidrGroupsCache", // AllowAllCIDRAuthorizer doesn't support this operation, as feature is disabled by default
"org.apache.cassandra.db:type=StorageService:forceRemoveCompletion" // deprecated (TCM)
);