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

Port GossipTest::nodeDownDuringMove to 4.0 

Patch by Sam Tunnicliffe; reviewed by Ekaterina Dimitrova for CASSANDRA-15148
This commit is contained in:
Sam Tunnicliffe 2021-03-18 10:49:01 +00:00
parent d656f8ac01
commit bf96367f4d
2 changed files with 90 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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;

87
test/distributed/org/apache/cassandra/distributed/test/GossipTest.java
View File

@ -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);