mirror of https://github.com/apache/cassandra
Expose bootstrap and decommission state to nodetool info
patch by Stefan Miklosovic; reviewed by Brandon Williams CASSANDRA-18555 Co-authored-by: Jaydeepkumar Chovatia <chovatia.jaydeep@gmail.com>
This commit is contained in:
Stefan Miklosovic
parent
f5a59176f9
commit
e2a6c99310
|
|
@ -1,4 +1,5 @@
|
|||
5.0
|
||||
* Expose bootstrap and decommission state to nodetool info (CASSANDRA-18555)
|
||||
* Fix SSTabledump errors when dumping data from index (CASSANDRA-17698)
|
||||
* Avoid unnecessary deserialization of terminal arguments when executing CQL functions (CASSANDRA-18566)
|
||||
* Remove dependency on pytz library for setting CQLSH timezones on Python version >= 3.9 (CASSANDRA-17433)
|
||||
|
|
|
|||
1
NEWS.txt
1
NEWS.txt
|
|
@ -164,6 +164,7 @@ New features
|
|||
- Added `sstablepartitions` offline tool to find large partitions in sstables.
|
||||
- `cassandra-stress` has a new option called '-jmx' which enables a user to pass username and password to JMX (CASSANDRA-18544)
|
||||
- It is possible to read all credentials for `cassandra-stress` from a file via option `-credentials-file` (CASSANDRA-18544)
|
||||
- nodetool info displays bootstrap state a node is in as well as if it was decommissioned or if it failed to decommission (CASSANDRA-18555)
|
||||
|
||||
Upgrading
|
||||
---------
|
||||
|
|
|
|||
|
|
@ -263,6 +263,8 @@ import static org.apache.cassandra.net.NoPayload.noPayload;
|
|||
import static org.apache.cassandra.net.Verb.REPLICATION_DONE_REQ;
|
||||
import static org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus;
|
||||
import static org.apache.cassandra.service.ActiveRepairService.repairCommandExecutor;
|
||||
import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSIONED;
|
||||
import static org.apache.cassandra.service.StorageService.Mode.DECOMMISSION_FAILED;
|
||||
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
|
||||
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
|
||||
import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
|
||||
|
|
@ -414,7 +416,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, LEAVING, DECOMMISSIONED, MOVING, DRAINING, DRAINED }
|
||||
public enum Mode { STARTING, NORMAL, JOINING, LEAVING, DECOMMISSIONED, DECOMMISSION_FAILED, MOVING, DRAINING, DRAINED }
|
||||
private volatile Mode operationMode = Mode.STARTING;
|
||||
|
||||
/* Used for tracking drain progress */
|
||||
|
|
@ -626,7 +628,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
* they get the Gossip shutdown message, so even if
|
||||
* we don't get time to broadcast this, it is not a problem.
|
||||
*
|
||||
* See {@link Gossiper#markAsShutdown(InetAddressAndPort)}
|
||||
* See Gossiper.markAsShutdown(InetAddressAndPort)
|
||||
*/
|
||||
private void shutdownClientServers()
|
||||
{
|
||||
|
|
@ -2157,7 +2159,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
/**
|
||||
* All MVs have been created during bootstrap, so mark them as built
|
||||
*/
|
||||
private void markViewsAsBuilt() {
|
||||
private void markViewsAsBuilt()
|
||||
{
|
||||
for (String keyspace : Schema.instance.getUserKeyspaces().names())
|
||||
{
|
||||
for (ViewMetadata view: Schema.instance.getKeyspaceMetadata(keyspace).views)
|
||||
|
|
@ -2168,11 +2171,18 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
/**
|
||||
* Called when bootstrap did finish successfully
|
||||
*/
|
||||
private void bootstrapFinished() {
|
||||
private void bootstrapFinished()
|
||||
{
|
||||
markViewsAsBuilt();
|
||||
isBootstrapMode = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getBootstrapState()
|
||||
{
|
||||
return SystemKeyspace.getBootstrapState().name();
|
||||
}
|
||||
|
||||
public boolean resumeBootstrap()
|
||||
{
|
||||
if (isBootstrapMode && SystemKeyspace.bootstrapInProgress())
|
||||
|
|
@ -5128,18 +5138,32 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
|
||||
public void decommission(boolean force) throws InterruptedException
|
||||
{
|
||||
if (operationMode == DECOMMISSIONED)
|
||||
{
|
||||
logger.info("This node was already decommissioned. There is no point in decommissioning it again.");
|
||||
return;
|
||||
}
|
||||
|
||||
if (isDecommissioning())
|
||||
{
|
||||
logger.info("This node is still decommissioning.");
|
||||
return;
|
||||
}
|
||||
|
||||
TokenMetadata metadata = tokenMetadata.cloneAfterAllLeft();
|
||||
// there is no point to do this logic again once node was decommissioning but failed to do so
|
||||
if (operationMode != Mode.LEAVING)
|
||||
{
|
||||
if (!tokenMetadata.isMember(FBUtilities.getBroadcastAddressAndPort()))
|
||||
throw new UnsupportedOperationException("local node is not a member of the token ring yet");
|
||||
if (metadata.getAllEndpoints().size() < 2)
|
||||
if (metadata.getAllEndpoints().size() < 2 && metadata.getAllEndpoints().contains(FBUtilities.getBroadcastAddressAndPort()))
|
||||
throw new UnsupportedOperationException("no other normal nodes in the ring; decommission would be pointless");
|
||||
if (operationMode != Mode.NORMAL)
|
||||
if (operationMode != Mode.NORMAL && operationMode != DECOMMISSION_FAILED)
|
||||
throw new UnsupportedOperationException("Node in " + operationMode + " state; wait for status to become normal or restart");
|
||||
}
|
||||
|
||||
if (!isDecommissioning.compareAndSet(false, true))
|
||||
throw new IllegalStateException("Node is still decommissioning. Check nodetool netstats.");
|
||||
throw new IllegalStateException("Node is still decommissioning. Check nodetool netstats or nodetool info.");
|
||||
|
||||
if (logger.isDebugEnabled())
|
||||
logger.debug("DECOMMISSIONING");
|
||||
|
|
@ -5150,27 +5174,35 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
|
||||
String dc = DatabaseDescriptor.getEndpointSnitch().getLocalDatacenter();
|
||||
|
||||
if (operationMode != Mode.LEAVING) // If we're already decommissioning there is no point checking RF/pending ranges
|
||||
// If we're already decommissioning there is no point checking RF/pending ranges
|
||||
if (operationMode != Mode.LEAVING)
|
||||
{
|
||||
int rf, numNodes;
|
||||
for (String keyspaceName : Schema.instance.getNonLocalStrategyKeyspaces().names())
|
||||
{
|
||||
if (!force)
|
||||
{
|
||||
boolean notEnoughLiveNodes = false;
|
||||
Keyspace keyspace = Keyspace.open(keyspaceName);
|
||||
if (keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy)
|
||||
{
|
||||
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
|
||||
rf = strategy.getReplicationFactor(dc).allReplicas;
|
||||
numNodes = metadata.getTopology().getDatacenterEndpoints().get(dc).size();
|
||||
Collection<InetAddressAndPort> datacenterEndpoints = metadata.getTopology().getDatacenterEndpoints().get(dc);
|
||||
numNodes = datacenterEndpoints.size();
|
||||
if (numNodes <= rf && datacenterEndpoints.contains(FBUtilities.getBroadcastAddressAndPort()))
|
||||
notEnoughLiveNodes = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
numNodes = metadata.getAllEndpoints().size();
|
||||
Set<InetAddressAndPort> allEndpoints = metadata.getAllEndpoints();
|
||||
numNodes = allEndpoints.size();
|
||||
rf = keyspace.getReplicationStrategy().getReplicationFactor().allReplicas;
|
||||
if (numNodes <= rf && allEndpoints.contains(FBUtilities.getBroadcastAddressAndPort()))
|
||||
notEnoughLiveNodes = true;
|
||||
}
|
||||
|
||||
if (numNodes <= rf)
|
||||
if (notEnoughLiveNodes)
|
||||
throw new UnsupportedOperationException("Not enough live nodes to maintain replication factor in keyspace "
|
||||
+ keyspaceName + " (RF = " + rf + ", N = " + numNodes + ")."
|
||||
+ " Perform a forceful decommission to ignore.");
|
||||
|
|
@ -5182,42 +5214,48 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
}
|
||||
|
||||
startLeaving();
|
||||
long timeout = Math.max(RING_DELAY_MILLIS, BatchlogManager.instance.getBatchlogTimeout());
|
||||
long timeout = Math.max(RING_DELAY_MILLIS, BatchlogManager.getBatchlogTimeout());
|
||||
setMode(Mode.LEAVING, "sleeping " + timeout + " ms for batch processing and pending range setup", true);
|
||||
Thread.sleep(timeout);
|
||||
|
||||
Runnable finishLeaving = new Runnable()
|
||||
{
|
||||
public void run()
|
||||
{
|
||||
shutdownClientServers();
|
||||
Gossiper.instance.stop();
|
||||
try
|
||||
{
|
||||
MessagingService.instance().shutdown();
|
||||
}
|
||||
catch (IOError ioe)
|
||||
{
|
||||
logger.info("failed to shutdown message service: {}", ioe);
|
||||
}
|
||||
unbootstrap();
|
||||
|
||||
Stage.shutdownNow();
|
||||
SystemKeyspace.setBootstrapState(SystemKeyspace.BootstrapState.DECOMMISSIONED);
|
||||
setMode(Mode.DECOMMISSIONED, true);
|
||||
// let op be responsible for killing the process
|
||||
}
|
||||
};
|
||||
unbootstrap(finishLeaving);
|
||||
// shutdown cql, gossip, messaging, Stage and set state to DECOMMISSIONED
|
||||
|
||||
shutdownClientServers();
|
||||
Gossiper.instance.stop();
|
||||
try
|
||||
{
|
||||
MessagingService.instance().shutdown();
|
||||
}
|
||||
catch (IOError ioe)
|
||||
{
|
||||
logger.info("failed to shutdown message service", ioe);
|
||||
}
|
||||
|
||||
Stage.shutdownNow();
|
||||
SystemKeyspace.setBootstrapState(SystemKeyspace.BootstrapState.DECOMMISSIONED);
|
||||
setMode(DECOMMISSIONED, true);
|
||||
// let op be responsible for killing the process
|
||||
}
|
||||
catch (InterruptedException e)
|
||||
{
|
||||
throw new UncheckedInterruptedException(e);
|
||||
setMode(DECOMMISSION_FAILED, true);
|
||||
logger.error("Node interrupted while decommissioning");
|
||||
throw new RuntimeException("Node interrupted while decommissioning");
|
||||
}
|
||||
catch (ExecutionException e)
|
||||
{
|
||||
logger.error("Error while decommissioning node ", e.getCause());
|
||||
setMode(DECOMMISSION_FAILED, true);
|
||||
logger.error("Error while decommissioning node: {}", e.getCause().getMessage());
|
||||
throw new RuntimeException("Error while decommissioning node: " + e.getCause().getMessage());
|
||||
}
|
||||
catch (Throwable t)
|
||||
{
|
||||
setMode(DECOMMISSION_FAILED, true);
|
||||
logger.error("Error while decommissioning node: {}", t.getMessage());
|
||||
throw t;
|
||||
}
|
||||
finally
|
||||
{
|
||||
isDecommissioning.set(false);
|
||||
|
|
@ -5254,7 +5292,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
return () -> streamRanges(rangesToStream);
|
||||
}
|
||||
|
||||
private void unbootstrap(Runnable onFinish) throws ExecutionException, InterruptedException
|
||||
private void unbootstrap() throws ExecutionException, InterruptedException
|
||||
{
|
||||
Supplier<Future<StreamState>> startStreaming = prepareUnbootstrapStreaming();
|
||||
|
||||
|
|
@ -5290,7 +5328,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
hintsSuccess.get();
|
||||
logger.debug("stream acks all received.");
|
||||
leaveRing();
|
||||
onFinish.run();
|
||||
}
|
||||
|
||||
private Future streamHints()
|
||||
|
|
@ -5610,7 +5647,17 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
|
||||
public boolean isDecommissioned()
|
||||
{
|
||||
return operationMode == Mode.DECOMMISSIONED;
|
||||
return operationMode == DECOMMISSIONED;
|
||||
}
|
||||
|
||||
public boolean isDecommissionFailed()
|
||||
{
|
||||
return operationMode == DECOMMISSION_FAILED;
|
||||
}
|
||||
|
||||
public boolean isDecommissioning()
|
||||
{
|
||||
return isDecommissioning.get();
|
||||
}
|
||||
|
||||
public String getDrainProgress()
|
||||
|
|
|
|||
|
|
@ -489,6 +489,23 @@ public interface StorageServiceMBean extends NotificationEmitter
|
|||
*/
|
||||
public void decommission(boolean force) throws InterruptedException;
|
||||
|
||||
/**
|
||||
* Returns whether a node has failed to decommission.
|
||||
*
|
||||
* The fact that this method returns false does not mean that there was an attempt to
|
||||
* decommission this node which was successful.
|
||||
*
|
||||
* @return true if decommission of this node has failed, false otherwise
|
||||
*/
|
||||
public boolean isDecommissionFailed();
|
||||
|
||||
/**
|
||||
* Returns whether a node is being decommissioned or not.
|
||||
*
|
||||
* @return true if this node is decommissioning, false otherwise
|
||||
*/
|
||||
public boolean isDecommissioning();
|
||||
|
||||
/**
|
||||
* @param newToken token to move this node to.
|
||||
* This node will unload its data onto its neighbors, and bootstrap to the new token.
|
||||
|
|
@ -1006,6 +1023,8 @@ public interface StorageServiceMBean extends NotificationEmitter
|
|||
*/
|
||||
public boolean resumeBootstrap();
|
||||
|
||||
public String getBootstrapState();
|
||||
|
||||
/** Gets the concurrency settings for processing stages*/
|
||||
static class StageConcurrency implements Serializable
|
||||
{
|
||||
|
|
|
|||
|
|
@ -37,6 +37,16 @@ public class Decommission extends NodeToolCmd
|
|||
{
|
||||
try
|
||||
{
|
||||
if (probe.getStorageService().isDecommissioning())
|
||||
{
|
||||
probe.output().out.println("This node is still decommissioning.");
|
||||
return;
|
||||
}
|
||||
if ("DECOMMISSIONED".equals(probe.getStorageService().getBootstrapState()))
|
||||
{
|
||||
probe.output().out.println("Node was already decommissioned.");
|
||||
return;
|
||||
}
|
||||
probe.decommission(force);
|
||||
} catch (InterruptedException e)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -177,6 +177,10 @@ public class Info extends NodeToolCmd
|
|||
{
|
||||
out.printf("%-23s: (node is not joined to the cluster)%n", "Token");
|
||||
}
|
||||
|
||||
out.printf("%-23s: %s%n", "Bootstrap state", probe.getStorageService().getBootstrapState());
|
||||
out.printf("%-23s: %s%n", "Decommissioning", probe.getStorageService().isDecommissioning());
|
||||
out.printf("%-23s: %s%n", "Decommission failed", probe.getStorageService().isDecommissionFailed());
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -0,0 +1,220 @@
|
|||
/*
|
||||
* 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.util.concurrent.Callable;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
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.IInvokableInstance;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.streaming.StreamState;
|
||||
import org.apache.cassandra.utils.concurrent.Future;
|
||||
|
||||
import static net.bytebuddy.matcher.ElementMatchers.named;
|
||||
import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.COMPLETED;
|
||||
import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.DECOMMISSIONED;
|
||||
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;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
public class DecommissionTest extends TestBaseImpl
|
||||
{
|
||||
@Test
|
||||
public void testDecommission() throws Throwable
|
||||
{
|
||||
try (Cluster cluster = init(Cluster.build(2)
|
||||
.withConfig(config -> config.with(GOSSIP)
|
||||
.with(NETWORK))
|
||||
.withInstanceInitializer(DecommissionTest.BB::install)
|
||||
.start()))
|
||||
{
|
||||
IInvokableInstance instance = cluster.get(1);
|
||||
|
||||
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)
|
||||
{
|
||||
assertEquals("simulated error in prepareUnbootstrapStreaming", t.getMessage());
|
||||
}
|
||||
|
||||
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());
|
||||
}
|
||||
|
||||
// check that decommissioning of already decommissioned node has no effect
|
||||
|
||||
try
|
||||
{
|
||||
assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
|
||||
assertFalse(StorageService.instance.isDecommissionFailed());
|
||||
|
||||
StorageService.instance.decommission(true);
|
||||
|
||||
assertEquals(DECOMMISSIONED.name(), StorageService.instance.getBootstrapState());
|
||||
assertFalse(StorageService.instance.isDecommissionFailed());
|
||||
assertFalse(StorageService.instance.isDecommissioning());
|
||||
}
|
||||
catch (Throwable t)
|
||||
{
|
||||
fail("Decommissioning already decommissioned node should be no-op operation.");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDecommissionAfterNodeRestart() throws Throwable
|
||||
{
|
||||
try (Cluster cluster = init(Cluster.build(2)
|
||||
.withConfig(config -> config.with(GOSSIP)
|
||||
.with(NETWORK))
|
||||
.withInstanceInitializer((classLoader, threadGroup, num, generation) -> {
|
||||
// we do not want to install BB after restart of a node which
|
||||
// failed to decommission which is the second generation, here
|
||||
// as "1" as it is counted from 0.
|
||||
if (num == 1 && generation != 1)
|
||||
BB.install(classLoader, num);
|
||||
})
|
||||
.start()))
|
||||
{
|
||||
IInvokableInstance instance = cluster.get(1);
|
||||
|
||||
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)
|
||||
{
|
||||
assertEquals("simulated error in prepareUnbootstrapStreaming", t.getMessage());
|
||||
}
|
||||
|
||||
// node is in DECOMMISSION_FAILED mode
|
||||
String operationMode = StorageService.instance.getOperationMode();
|
||||
assertEquals(DECOMMISSION_FAILED.name(), operationMode);
|
||||
});
|
||||
|
||||
// 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);
|
||||
|
||||
instance.runOnInstance(() -> {
|
||||
try
|
||||
{
|
||||
StorageService.instance.decommission(true);
|
||||
}
|
||||
catch (InterruptedException e)
|
||||
{
|
||||
fail("Should decommission the node");
|
||||
}
|
||||
|
||||
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)
|
||||
{
|
||||
new ByteBuddy().rebase(StorageService.class)
|
||||
.method(named("prepareUnbootstrapStreaming"))
|
||||
.intercept(MethodDelegation.to(DecommissionTest.BB.class))
|
||||
.make()
|
||||
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
|
||||
}
|
||||
|
||||
private static int invocations = 0;
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
public static Supplier<Future<StreamState>> prepareUnbootstrapStreaming(@SuperCall Callable<Supplier<Future<StreamState>>> zuper)
|
||||
{
|
||||
++invocations;
|
||||
|
||||
if (invocations == 1)
|
||||
throw new RuntimeException("simulated error in prepareUnbootstrapStreaming");
|
||||
|
||||
try
|
||||
{
|
||||
return zuper.call();
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue