CASSANDRA-18866-5.0 patch

Author: smiklosovic

src/java/org/apache/cassandra/gms/Gossiper.java

@@ -55,6 +55,7 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.Uninterruptibles;
+import org.apache.cassandra.exceptions.RequestFailureReason;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -165,6 +166,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
     @VisibleForTesting
     final Set<InetAddressAndPort> liveEndpoints = new ConcurrentSkipListSet<>();
 
+    /* Inflight echo requests. */
+    private final Set<InetAddressAndPort> inflightEcho = new ConcurrentSkipListSet<>();
+
     /* unreachable member set */
     private final Map<InetAddressAndPort, Long> unreachableEndpoints = new ConcurrentHashMap<>();
 
@@ -729,6 +733,7 @@ public void removeEndpoint(InetAddressAndPort endpoint)
             return;
 
         liveEndpoints.remove(endpoint);
+        inflightEcho.remove(endpoint);
         unreachableEndpoints.remove(endpoint);
         MessagingService.instance().versions.reset(endpoint);
         quarantineEndpoint(endpoint);
@@ -1430,12 +1435,53 @@ void notifyFailureDetector(InetAddressAndPort endpoint, EndpointState remoteEndp
     private void markAlive(final InetAddressAndPort addr, final EndpointState localState)
     {
         localState.markDead();
+        if (!inflightEcho.add(addr))
+        {
+            return;
+        }
 
         Message<NoPayload> echoMessage = Message.out(ECHO_REQ, noPayload);
         logger.trace("Sending ECHO_REQ to {}", addr);
-        RequestCallback echoHandler = msg ->
+        RequestCallback echoHandler = new RequestCallback()
         {
-            runInGossipStageBlocking(() -> realMarkAlive(addr, localState));
+            @Override
+            public void onResponse(Message msg)
+            {
+                // force processing of the echo response onto the gossip stage, as it comes in on the REQUEST_RESPONSE stage
+                runInGossipStageBlocking(() -> {
+                    try
+                    {
+                        EndpointState localStatePtr = endpointStateMap.get(addr);
+                        realMarkAlive(addr, localStatePtr);
+                    }
+                    finally
+                    {
+                        inflightEcho.remove(addr);
+                    }
+                });
+            }
+
+            @Override
+            public boolean invokeOnFailure()
+            {
+                return true;
+            }
+
+            @Override
+            public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason)
+            {
+                if (isEnabled())
+                {
+                    logger.trace("Resending ECHO_REQ to {}", addr);
+                    Message<NoPayload> echoMessage = Message.out(ECHO_REQ, noPayload);
+                    MessagingService.instance().sendWithCallback(echoMessage, addr, this);
+                }
+                else
+                {
+                    logger.trace("Failed ECHO_REQ to {}, aborting due to disabled gossip", addr);
+                    inflightEcho.remove(addr);
+                }
+            }
         };
 
         MessagingService.instance().sendWithCallback(echoMessage, addr, echoHandler);
@@ -1488,6 +1534,7 @@ public void markDead(InetAddressAndPort addr, EndpointState localState)
     private void silentlyMarkDead(InetAddressAndPort addr, EndpointState localState)
     {
         localState.markDead();
+        inflightEcho.remove(addr);
         if (!disableEndpointRemoval)
         {
             liveEndpoints.remove(addr);