Repository: cassandra Updated Branches: refs/heads/cassandra-2.0 35a4f7dd6 -> 08dbbd689 refs/heads/cassandra-2.1 7043a66d0 -> ea1beda01 refs/heads/trunk cc5128a02 -> d4f2354e4
Ignore gossip SYNs after shutdown Patch by Sergio Bossa, reviewed by brandonwilliams for CASSANDRA-9238 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/08dbbd68 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/08dbbd68 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/08dbbd68 Branch: refs/heads/cassandra-2.0 Commit: 08dbbd689ee7db7443c6920dfcd402710c1dd964 Parents: 35a4f7d Author: Brandon Williams <brandonwilli...@apache.org> Authored: Mon Apr 27 14:37:33 2015 -0500 Committer: Brandon Williams <brandonwilli...@apache.org> Committed: Mon Apr 27 14:37:33 2015 -0500 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../gms/GossipDigestSynVerbHandler.java | 8 ++++++ src/java/org/apache/cassandra/gms/Gossiper.java | 26 ++++++++++++++++++++ 3 files changed, 35 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/08dbbd68/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 2613901..c54bc45 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.0.15: + * Ignore gossip SYNs after shutdown (CASSANDRA-9238) * Avoid overflow when calculating max sstable size in LCS (CASSANDRA-9235) * Make sstable blacklisting work with compression (CASSANDRA-9138) * Do not attempt to rebuild indexes if no index accepts any column (CASSANDRA-9196) http://git-wip-us.apache.org/repos/asf/cassandra/blob/08dbbd68/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java index df74808..4454c46 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java @@ -36,14 +36,22 @@ public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> public void doVerb(MessageIn<GossipDigestSyn> message, int id) { InetAddress from = message.from; + if (logger.isTraceEnabled()) logger.trace("Received a GossipDigestSynMessage from {}", from); + if (!Gossiper.instance.isEnabled()) { if (logger.isTraceEnabled()) logger.trace("Ignoring GossipDigestSynMessage because gossip is disabled"); return; } + + if (!Gossiper.instance.shouldAckAfterShutdown(from)) + { + logger.debug("Temporarily ignoring SYN from shutdown node {}", from); + return; + } GossipDigestSyn gDigestMessage = message.payload; /* If the message is from a different cluster throw it away. */ http://git-wip-us.apache.org/repos/asf/cassandra/blob/08dbbd68/src/java/org/apache/cassandra/gms/Gossiper.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index b77064d..4665e74 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -103,6 +103,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /* unreachable member set */ private final Map<InetAddress, Long> unreachableEndpoints = new ConcurrentHashMap<InetAddress, Long>(); + + /* shutdown member set */ + private final Map<InetAddress, Long> shutdownEndpoints = new ConcurrentHashMap<InetAddress, Long>(); /* initial seeds for joining the cluster */ private final Set<InetAddress> seeds = new ConcurrentSkipListSet<InetAddress>(inetcomparator); @@ -302,6 +305,24 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return 0L; } + public boolean shouldAckAfterShutdown(InetAddress endpoint) + { + Long shutdownTimestamp = shutdownEndpoints.get(endpoint); + Integer shutdownAnnounce = Integer.getInteger("cassandra.shutdown_announce_in_ms", 2000); + // Do not temporarily answer to SYN messages coming from shutdown nodes, to avoid reconnecting: + if (shutdownTimestamp != null && (System.currentTimeMillis() - shutdownTimestamp) < (shutdownAnnounce * 2)) + { + return false; + } + // Otherwise, if allowed to answer, remove the node from the shutdown set + // (and do this only here, rather than in sendGossip too, to avoid racing between different calling threads): + else + { + shutdownEndpoints.remove(endpoint); + return true; + } + } + private boolean isShutdown(InetAddress endpoint) { EndpointState epState = endpointStateMap.get(endpoint); @@ -350,6 +371,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return; epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.shutdown(true)); epState.getHeartBeatState().forceHighestPossibleVersionUnsafe(); + shutdownEndpoints.put(endpoint, System.currentTimeMillis()); markDead(endpoint, epState); FailureDetector.instance.forceConviction(endpoint); } @@ -622,6 +644,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /* Generate a random number from 0 -> size */ int index = (size == 1) ? 0 : random.nextInt(size); InetAddress to = liveEndpoints.get(index); + Long shutdownTimestamp = shutdownEndpoints.get(to); + Integer shutdownAnnounce = Integer.getInteger("cassandra.shutdown_announce_in_ms", 2000); + if (shutdownTimestamp != null && (System.currentTimeMillis() - shutdownTimestamp) < (shutdownAnnounce * 2)) + return false; if (logger.isTraceEnabled()) logger.trace("Sending a GossipDigestSyn to {} ...", to); MessagingService.instance().sendOneWay(message, to);
