From bf96367f4d55692017e144980cf17963e31df127 Mon Sep 17 00:00:00 2001 From: Sam Tunnicliffe Date: Thu, 18 Mar 2021 10:49:01 +0000 Subject: [PATCH] Port GossipTest::nodeDownDuringMove to 4.0 Patch by Sam Tunnicliffe; reviewed by Ekaterina Dimitrova for CASSANDRA-15148 --- .../cassandra/distributed/impl/Instance.java | 3 + .../distributed/test/GossipTest.java | 87 +++++++++++++++++++ 2 files changed, 90 insertions(+) diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 4a865a990a..d772d51758 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -690,6 +690,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance @Override public Future shutdown(boolean graceful) { + if (!graceful) + MessagingService.instance().shutdown(1L, MINUTES, false, true); + Future future = async((ExecutorService executor) -> { Throwable error = null; diff --git a/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java b/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java index 1b6a0045d3..7ebfd0c3b3 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/GossipTest.java @@ -19,13 +19,17 @@ package org.apache.cassandra.distributed.test; import java.io.Closeable; +import java.net.InetSocketAddress; import java.util.Collection; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; +import java.util.stream.Collectors; +import com.google.common.collect.Iterables; import com.google.common.util.concurrent.Uninterruptibles; import org.junit.Assert; import org.junit.Test; @@ -35,15 +39,98 @@ import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; import net.bytebuddy.implementation.MethodDelegation; import org.apache.cassandra.dht.Token; import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.EndpointState; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.FBUtilities; import static net.bytebuddy.matcher.ElementMatchers.named; import static net.bytebuddy.matcher.ElementMatchers.takesArguments; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.impl.DistributedTestSnitch.toCassandraInetAddressAndPort; public class GossipTest extends TestBaseImpl { + @Test + public void nodeDownDuringMove() throws Throwable + { + int liveCount = 1; + try (Cluster cluster = Cluster.build(2 + liveCount) + .withConfig(config -> config.with(NETWORK).with(GOSSIP)) + .createWithoutStarting()) + { + int fail = liveCount + 1; + int late = fail + 1; + for (int i = 1 ; i <= liveCount ; ++i) + cluster.get(i).startup(); + cluster.get(fail).startup(); + Collection expectTokens = + cluster.get(fail) + .callsOnInstance(() -> StorageService.instance.getTokenMetadata() + .getTokens(FBUtilities.getBroadcastAddressAndPort()) + .stream() + .map(Object::toString) + .collect(Collectors.toList())) + .call(); + + InetSocketAddress failAddress = cluster.get(fail).broadcastAddress(); + // wait for NORMAL state + for (int i = 1 ; i <= liveCount ; ++i) + { + cluster.get(i).acceptsOnInstance((InetSocketAddress address) -> { + EndpointState ep; + InetAddressAndPort endpoint = toCassandraInetAddressAndPort(address); + while (null == (ep = Gossiper.instance.getEndpointStateForEndpoint(endpoint)) + || ep.getApplicationState(ApplicationState.STATUS_WITH_PORT) == null + || !ep.getApplicationState(ApplicationState.STATUS_WITH_PORT).value.startsWith("NORMAL")) + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10L)); + }).accept(failAddress); + } + + // set ourselves to MOVING, and wait for it to propagate + cluster.get(fail).runOnInstance(() -> { + Token token = Iterables.getFirst(StorageService.instance.getTokenMetadata().getTokens(FBUtilities.getBroadcastAddressAndPort()), null); + Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.moving(token)); + }); + for (int i = 1 ; i <= liveCount ; ++i) + { + cluster.get(i).acceptsOnInstance((InetSocketAddress address) -> { + EndpointState ep; + InetAddressAndPort endpoint = toCassandraInetAddressAndPort(address); + while (null == (ep = Gossiper.instance.getEndpointStateForEndpoint(endpoint)) + || (ep.getApplicationState(ApplicationState.STATUS_WITH_PORT) == null + || !ep.getApplicationState(ApplicationState.STATUS_WITH_PORT).value.startsWith("MOVING"))) + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100L)); + }).accept(failAddress); + } + + cluster.get(fail).shutdown(false).get(); + cluster.get(late).startup(); + cluster.get(late).acceptsOnInstance((InetSocketAddress address) -> { + EndpointState ep; + InetAddressAndPort endpoint = toCassandraInetAddressAndPort(address); + while (null == (ep = Gossiper.instance.getEndpointStateForEndpoint(endpoint)) + || !ep.getApplicationState(ApplicationState.STATUS_WITH_PORT).value.startsWith("MOVING")) + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100L)); + }).accept(failAddress); + + Collection tokens = + cluster.get(late) + .appliesOnInstance((InetSocketAddress address) -> + StorageService.instance.getTokenMetadata() + .getTokens(toCassandraInetAddressAndPort(address)) + .stream() + .map(Object::toString) + .collect(Collectors.toList())) + .apply(failAddress); + + Assert.assertEquals(expectTokens, tokens); + } + } + public static class BBBootstrapInterceptor { final static CountDownLatch bootstrapReady = new CountDownLatch(1);