From bf0906b92cf65161d828e31bc46436d427bbb4b8 Mon Sep 17 00:00:00 2001 From: Stefan Podkowinski Date: Mon, 19 Sep 2016 13:56:54 +0200 Subject: [PATCH] Discard in-flight shadow round responses patch by Stefan Podkowinski; reviewed by Joel Knighton and Jason Brown for CASSANDRA-12653 --- CHANGES.txt | 1 + .../gms/GossipDigestAckVerbHandler.java | 26 +++++--- .../org/apache/cassandra/gms/Gossiper.java | 62 ++++++++++++++----- .../cassandra/service/MigrationTask.java | 12 ++-- .../cassandra/service/StorageService.java | 16 +++-- 5 files changed, 79 insertions(+), 38 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 27dd343905..df2421ddc1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.2.10 + * Discard in-flight shadow round responses (CASSANDRA-12653) * Don't anti-compact repaired data to avoid inconsistencies (CASSANDRA-13153) * Wrong logger name in AnticompactionTask (CASSANDRA-13343) * Fix queries updating multiple time the same list (CASSANDRA-13130) diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java index 9f69a945ef..59060f89bb 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java @@ -51,21 +51,31 @@ public class GossipDigestAckVerbHandler implements IVerbHandler Map epStateMap = gDigestAckMessage.getEndpointStateMap(); logger.trace("Received ack with {} digests and {} states", gDigestList.size(), epStateMap.size()); - if (epStateMap.size() > 0) - { - /* Notify the Failure Detector */ - Gossiper.instance.notifyFailureDetector(epStateMap); - Gossiper.instance.applyStateLocally(epStateMap); - } - if (Gossiper.instance.isInShadowRound()) { if (logger.isDebugEnabled()) logger.debug("Finishing shadow round with {}", from); - Gossiper.instance.finishShadowRound(); + Gossiper.instance.finishShadowRound(epStateMap); return; // don't bother doing anything else, we have what we came for } + if (epStateMap.size() > 0) + { + // Ignore any GossipDigestAck messages that we handle before a regular GossipDigestSyn has been send. + // This will prevent Acks from leaking over from the shadow round that are not actual part of + // the regular gossip conversation. + if ((System.nanoTime() - Gossiper.instance.firstSynSendAt) < 0 || Gossiper.instance.firstSynSendAt == 0) + { + if (logger.isTraceEnabled()) + logger.trace("Ignoring unrequested GossipDigestAck from {}", from); + return; + } + + /* Notify the Failure Detector */ + Gossiper.instance.notifyFailureDetector(epStateMap); + Gossiper.instance.applyStateLocally(epStateMap); + } + /* Get the state required to send to this gossipee - construct GossipDigestAck2Message */ Map deltaEpStateMap = new HashMap(); for (GossipDigest gDigest : gDigestList) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 06b14c4fab..c2eccbaf1a 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -30,6 +30,7 @@ import javax.management.ObjectName; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.Uninterruptibles; import org.apache.cassandra.utils.Pair; @@ -86,6 +87,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private static final Logger logger = LoggerFactory.getLogger(Gossiper.class); public static final Gossiper instance = new Gossiper(); + // Timestamp to prevent processing any in-flight messages for we've not send any SYN yet, see CASSANDRA-12653. + volatile long firstSynSendAt = 0L; + public static final long aVeryLongTime = 259200 * 1000; // 3 days // Maximimum difference between generation value and local time we are willing to accept about a peer @@ -125,6 +129,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private volatile boolean inShadowRound = false; + // endpoint states as gathered during shadow round + private final Map endpointShadowStateMap = new ConcurrentHashMap<>(); + private volatile long lastProcessedMessageAt = System.currentTimeMillis(); private class GossipTask implements Runnable @@ -645,6 +652,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean InetAddress to = liveEndpoints.get(index); if (logger.isTraceEnabled()) logger.trace("Sending a GossipDigestSyn to {} ...", to); + if (firstSynSendAt == 0) + firstSynSendAt = System.nanoTime(); MessagingService.instance().sendOneWay(message, to); return seeds.contains(to); } @@ -713,11 +722,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * Check if this endpoint can safely bootstrap into the cluster. * * @param endpoint - the endpoint to check + * @param epStates - endpoint states in the cluster * @return true if the endpoint can join the cluster */ - public boolean isSafeForBootstrap(InetAddress endpoint) + public boolean isSafeForBootstrap(InetAddress endpoint, Map epStates) { - EndpointState epState = endpointStateMap.get(endpoint); + EndpointState epState = epStates.get(endpoint); // if there's no previous state, or the node was previously removed from the cluster, we're good if (epState == null || isDeadState(epState)) @@ -816,14 +826,6 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return endpointStateMap.get(ep); } - // removes ALL endpoint states; should only be called after shadow gossip - public void resetEndpointStateMap() - { - endpointStateMap.clear(); - unreachableEndpoints.clear(); - liveEndpoints.clear(); - } - public Set> getEndpointStates() { return endpointStateMap.entrySet(); @@ -831,7 +833,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public UUID getHostId(InetAddress endpoint) { - return UUID.fromString(getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.HOST_ID).value); + return getHostId(endpoint, endpointStateMap); + } + + public UUID getHostId(InetAddress endpoint, Map epStates) + { + return UUID.fromString(epStates.get(endpoint).getApplicationState(ApplicationState.HOST_ID).value); } EndpointState getStateForVersionBiggerThan(InetAddress forEndpoint, int version) @@ -1305,12 +1312,32 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } /** - * Do a single 'shadow' round of gossip, where we do not modify any state - * Only used when replacing a node, to get and assume its states + * Do a single 'shadow' round of gossip by retrieving endpoint states that will be stored exclusively in the + * map return value, instead of endpointStateMap. + * + * Used when preparing to join the ring: + *
    + *
  • when replacing a node, to get and assume its tokens
    • + *
  • when joining, to check that the local host id matches any previous id for the endpoint address
    • + *
+ * + * Method is synchronized, as we use an in-progress flag to indicate that shadow round must be cleared + * again by calling {@link Gossiper#finishShadowRound(Map)}. This will update + * {@link Gossiper#endpointShadowStateMap} with received values, in order to return an immutable copy to the + * caller of {@link Gossiper#doShadowRound()}. Therefor only a single shadow round execution is permitted at + * the same time. + * + * @return endpoint states gathered during shadow round or empty map */ - public void doShadowRound() + public synchronized Map doShadowRound() { buildSeedsList(); + // it may be that the local address is the only entry in the seed + // list in which case, attempting a shadow round is pointless + if (seeds.isEmpty()) + return endpointShadowStateMap; + + endpointShadowStateMap.clear(); // send a completely empty syn List gDigests = new ArrayList(); GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(), @@ -1346,6 +1373,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { throw new RuntimeException(wtf); } + + return ImmutableMap.copyOf(endpointShadowStateMap); } private void buildSeedsList() @@ -1466,10 +1495,13 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return (scheduledGossipTask != null) && (!scheduledGossipTask.isCancelled()); } - protected void finishShadowRound() + protected void finishShadowRound(Map epStateMap) { if (inShadowRound) + { + endpointShadowStateMap.putAll(epStateMap); inShadowRound = false; + } } protected boolean isInShadowRound() diff --git a/src/java/org/apache/cassandra/service/MigrationTask.java b/src/java/org/apache/cassandra/service/MigrationTask.java index df0b76751e..b065d9093e 100644 --- a/src/java/org/apache/cassandra/service/MigrationTask.java +++ b/src/java/org/apache/cassandra/service/MigrationTask.java @@ -48,6 +48,12 @@ class MigrationTask extends WrappedRunnable public void runMayThrow() throws Exception { + if (!FailureDetector.instance.isAlive(endpoint)) + { + logger.warn("Can't send schema pull request: node {} is down.", endpoint); + return; + } + // There is a chance that quite some time could have passed between now and the MM#maybeScheduleSchemaPull(), // potentially enough for the endpoint node to restart - which is an issue if it does restart upgraded, with // a higher major. @@ -57,12 +63,6 @@ class MigrationTask extends WrappedRunnable return; } - if (!FailureDetector.instance.isAlive(endpoint)) - { - logger.debug("Can't send schema pull request: node {} is down.", endpoint); - return; - } - MessageOut message = new MessageOut<>(MessagingService.Verb.MIGRATION_REQUEST, null, MigrationManager.MigrationsSerializer.instance); IAsyncCallback> cb = new IAsyncCallback>() diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index c2996d7aeb..65f536b24d 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -443,15 +443,15 @@ public class StorageService extends NotificationBroadcasterSupport implements IE MessagingService.instance().listen(); // make magic happen - Gossiper.instance.doShadowRound(); + Map epStates = Gossiper.instance.doShadowRound(); // now that we've gossiped at least once, we should be able to find the node we're replacing - if (Gossiper.instance.getEndpointStateForEndpoint(DatabaseDescriptor.getReplaceAddress())== null) + if (epStates.get(DatabaseDescriptor.getReplaceAddress())== null) throw new RuntimeException("Cannot replace_address " + DatabaseDescriptor.getReplaceAddress() + " because it doesn't exist in gossip"); - replacingId = Gossiper.instance.getHostId(DatabaseDescriptor.getReplaceAddress()); + replacingId = Gossiper.instance.getHostId(DatabaseDescriptor.getReplaceAddress(), epStates); try { - VersionedValue tokensVersionedValue = Gossiper.instance.getEndpointStateForEndpoint(DatabaseDescriptor.getReplaceAddress()).getApplicationState(ApplicationState.TOKENS); + VersionedValue tokensVersionedValue = epStates.get(DatabaseDescriptor.getReplaceAddress()).getApplicationState(ApplicationState.TOKENS); if (tokensVersionedValue == null) throw new RuntimeException("Could not find tokens for " + DatabaseDescriptor.getReplaceAddress() + " to replace"); Collection tokens = TokenSerializer.deserialize(getPartitioner(), new DataInputStream(new ByteArrayInputStream(tokensVersionedValue.toBytes()))); @@ -460,7 +460,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE { SystemKeyspace.setLocalHostId(replacingId); // use the replacee's host Id as our own so we receive hints, etc } - Gossiper.instance.resetEndpointStateMap(); // clean up since we have what we need return tokens; } catch (IOException e) @@ -474,8 +473,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE logger.debug("Starting shadow gossip round to check for endpoint collision"); if (!MessagingService.instance().isListening()) MessagingService.instance().listen(); - Gossiper.instance.doShadowRound(); - if (!Gossiper.instance.isSafeForBootstrap(FBUtilities.getBroadcastAddress())) + Map epStates = Gossiper.instance.doShadowRound(); + if (!Gossiper.instance.isSafeForBootstrap(FBUtilities.getBroadcastAddress(), epStates)) { throw new RuntimeException(String.format("A node with address %s already exists, cancelling join. " + "Use cassandra.replace_address if you want to replace this node.", @@ -483,7 +482,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } if (useStrictConsistency && !allowSimultaneousMoves()) { - for (Map.Entry entry : Gossiper.instance.getEndpointStates()) + for (Map.Entry entry : epStates.entrySet()) { // ignore local node or empty status if (entry.getKey().equals(FBUtilities.getBroadcastAddress()) || entry.getValue().getApplicationState(ApplicationState.STATUS) == null) @@ -495,7 +494,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE throw new UnsupportedOperationException("Other bootstrapping/leaving/moving nodes detected, cannot bootstrap while cassandra.consistent.rangemovement is true"); } } - Gossiper.instance.resetEndpointStateMap(); } private boolean allowSimultaneousMoves()