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

Discard in-flight shadow round responses 

patch by Stefan Podkowinski; reviewed by Joel Knighton and Jason Brown for CASSANDRA-12653
This commit is contained in:
Stefan Podkowinski 2016-09-19 13:56:54 +02:00 committed by Joel Knighton
parent 06316df549
commit bf0906b92c
5 changed files with 79 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

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

26
src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java
View File

@ -51,21 +51,31 @@ public class GossipDigestAckVerbHandler implements IVerbHandler<GossipDigestAck>
Map<InetAddress, EndpointState> 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<InetAddress, EndpointState> deltaEpStateMap = new HashMap<InetAddress, EndpointState>();
for (GossipDigest gDigest : gDigestList)

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

@ -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<InetAddress, EndpointState> 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<InetAddress, EndpointState> 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<Entry<InetAddress, EndpointState>> 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<InetAddress, EndpointState> 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:
* <ul>
* <li>when replacing a node, to get and assume its tokens</li>
* <li>when joining, to check that the local host id matches any previous id for the endpoint address</li>
* </ul>
*
* 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<InetAddress, EndpointState> 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<GossipDigest> gDigests = new ArrayList<GossipDigest>();
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<InetAddress, EndpointState> epStateMap)
{
if (inShadowRound)
{
endpointShadowStateMap.putAll(epStateMap);
inShadowRound = false;
}
}
protected boolean isInShadowRound()

12
src/java/org/apache/cassandra/service/MigrationTask.java
View File

@ -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<Collection<Mutation>> cb = new IAsyncCallback<Collection<Mutation>>()

16
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -443,15 +443,15 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
MessagingService.instance().listen();
// make magic happen
Gossiper.instance.doShadowRound();
Map<InetAddress, EndpointState> 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<Token> 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<InetAddress, EndpointState> 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<InetAddress, EndpointState> entry : Gossiper.instance.getEndpointStates())
for (Map.Entry<InetAddress, EndpointState> 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()