This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new bf96367 Port GossipTest::nodeDownDuringMove to 4.0
bf96367 is described below
commit bf96367f4d55692017e144980cf17963e31df127
Author: Sam Tunnicliffe <[email protected]>
AuthorDate: Thu Mar 18 10:49:01 2021 +0000
Port GossipTest::nodeDownDuringMove to 4.0
Patch by Sam Tunnicliffe; reviewed by Ekaterina Dimitrova for
CASSANDRA-15148
---
.../cassandra/distributed/impl/Instance.java | 3 +
.../cassandra/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 4a865a9..d772d51 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<Void> 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 1b6a004..7ebfd0c 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<String> 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<String> 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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
