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

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Sam Tunnicliffe 2021-07-20 19:00:25 +01:00
parent 25dbbfddc2 fbb20b9162
commit b604bd20cd
3 changed files with 139 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -12,6 +12,7 @@
* Make sure sstables with moved starts are removed correctly in LeveledGenerations (CASSANDRA-16552)
* Upgrade jackson-databind to 2.9.10.8 (CASSANDRA-16462)
Merged from 3.0:
* Receipt of gossip shutdown notification updates TokenMetadata (CASSANDRA-16796)
* Count bloom filter misses correctly (CASSANDRA-12922)
* Reject token() in MV WHERE clause (CASSANDRA-13464)
* Ensure java executable is on the path (CASSANDRA-14325)

6
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -435,11 +435,15 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
EndpointState epState = endpointStateMap.get(endpoint);
if (epState == null)
return;
epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.shutdown(true));
VersionedValue shutdown = StorageService.instance.valueFactory.shutdown(true);
epState.addApplicationState(ApplicationState.STATUS, shutdown);
epState.addApplicationState(ApplicationState.RPC_READY, StorageService.instance.valueFactory.rpcReady(false));
epState.getHeartBeatState().forceHighestPossibleVersionUnsafe();
markDead(endpoint, epState);
FailureDetector.instance.forceConviction(endpoint);
for (IEndpointStateChangeSubscriber subscriber : subscribers)
subscriber.onChange(endpoint, ApplicationState.STATUS, shutdown);
logger.debug("Marked {} as shutdown", endpoint);
}
/**

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

@ -20,16 +20,15 @@ package org.apache.cassandra.distributed.test;
import java.io.Closeable;
import java.net.InetAddress;
import java.util.ArrayList;
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.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.Assert;
import org.junit.Test;
@ -39,16 +38,21 @@ 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.distributed.api.*;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.service.PendingRangeCalculatorService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamPlan;
import org.apache.cassandra.streaming.StreamResultFuture;
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.junit.Assert.assertEquals;
public class GossipTest extends TestBaseImpl
{
@ -224,4 +228,128 @@ public class GossipTest extends TestBaseImpl
}
}
@Test
public void gossipShutdownUpdatesTokenMetadata() throws Exception
{
try (Cluster cluster = Cluster.build(3)
.withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
.withInstanceInitializer(FailureHelper::installMoveFailure)
.start())
{
init(cluster, 2);
populate(cluster);
IInvokableInstance node1 = cluster.get(1);
IInvokableInstance node2 = cluster.get(2);
IInvokableInstance node3 = cluster.get(3);
// initiate a move for node2, which will not complete due to the
// ByteBuddy interceptor we injected. Wait for the other two nodes
// to mark node2 as moving before proceeding.
long t2 = Long.parseLong(getLocalToken(node2));
long t3 = Long.parseLong(getLocalToken(node3));
long moveTo = t2 + ((t3 - t2)/2);
String logMsg = "Node " + node2.broadcastAddress().getAddress() + " state moving, new token " + moveTo;
runAndWaitForLogs(() -> node2.nodetoolResult("move", "--", Long.toString(moveTo)).asserts().failure(),
logMsg,
cluster);
InetAddress movingAddress = node2.broadcastAddress().getAddress();
// node1 & node3 should now consider some ranges pending for node2
assertPendingRangesForPeer(true, movingAddress, cluster);
// A controlled shutdown causes peers to replace the MOVING status to be with SHUTDOWN, but prior to
// CASSANDRA-16796 this doesn't update TokenMetadata, so they maintain pending ranges for the down node
// indefinitely, even after it has been removed from the ring.
logMsg = "Marked " + node2.broadcastAddress().getAddress() + " as shutdown";
runAndWaitForLogs(() -> Futures.getUnchecked(node2.shutdown()),
logMsg,
node1, node3);
// node1 & node3 should not consider any ranges as still pending for node2
assertPendingRangesForPeer(false, movingAddress, cluster);
}
}
void assertPendingRangesForPeer(final boolean expectPending, final InetAddress movingAddress, final Cluster cluster)
{
for (IInvokableInstance inst : new IInvokableInstance[]{ cluster.get(1), cluster.get(3)})
{
boolean hasPending = inst.appliesOnInstance((InetAddress peer) -> {
PendingRangeCalculatorService.instance.blockUntilFinished();
boolean isMoving = StorageService.instance.getTokenMetadata()
.getMovingEndpoints()
.stream()
.map(pair -> pair.right)
.anyMatch(peer::equals);
return isMoving && !StorageService.instance.getTokenMetadata()
.getPendingRanges(KEYSPACE, peer)
.isEmpty();
}).apply(movingAddress);
assertEquals(String.format("%s should %shave PENDING RANGES for %s",
inst.broadcastAddress().getHostString(),
expectPending ? "" : "not ",
movingAddress),
hasPending, expectPending);
}
}
private String getLocalToken(IInvokableInstance inst)
{
return inst.callOnInstance(() -> {
List<String> tokens = new ArrayList<>();
for (Token t : StorageService.instance.getTokenMetadata().getTokens(FBUtilities.getBroadcastAddress()))
tokens.add(t.getTokenValue().toString());
assert tokens.size() == 1 : "getLocalToken assumes a single token, but multiple tokens found";
return tokens.get(0);
});
}
public static void runAndWaitForLogs(Runnable r, String waitString, Cluster cluster) throws TimeoutException
{
runAndWaitForLogs(r, waitString, cluster.stream().toArray(IInstance[]::new));
}
public static void runAndWaitForLogs(Runnable r, String waitString, IInstance...instances) throws TimeoutException
{
long [] marks = new long[instances.length];
for (int i = 0; i < instances.length; i++)
marks[i] = instances[i].logs().mark();
r.run();
for (int i = 0; i < instances.length; i++)
instances[i].logs().watchFor(marks[i], waitString);
}
static void populate(Cluster cluster)
{
cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl (pk int PRIMARY KEY)");
for (int i = 0; i < 10; i++)
{
cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk) VALUES (?)",
ConsistencyLevel.ALL,
i);
}
}
public static class FailureHelper
{
static void installMoveFailure(ClassLoader cl, int nodeNumber)
{
if (nodeNumber == 2)
{
new ByteBuddy().redefine(StreamPlan.class)
.method(named("execute"))
.intercept(MethodDelegation.to(FailureHelper.class))
.make()
.load(cl, ClassLoadingStrategy.Default.INJECTION);
}
}
public static StreamResultFuture execute()
{
throw new RuntimeException("failing to execute move");
}
}
}