CASSANDRA-20476 & CASSANDRA-20736 Handle CMS member addresses changing concurrently

src/java/org/apache/cassandra/config/Config.java

@@ -1581,6 +1581,7 @@ public static void log(Config config)
     public volatile DurationSpec.LongMillisecondsBound progress_barrier_timeout = new DurationSpec.LongMillisecondsBound("3600000ms");
     public volatile DurationSpec.LongMillisecondsBound progress_barrier_backoff = new DurationSpec.LongMillisecondsBound("1000ms");
     public volatile DurationSpec.LongSecondsBound discovery_timeout = new DurationSpec.LongSecondsBound("30s");
+    public volatile int discovery_rounds = 5;
     public boolean unsafe_tcm_mode = false;
     public boolean legacy_state_listener_sync_local_updates = true;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -6339,6 +6339,11 @@ public static long getDiscoveryTimeout(TimeUnit unit)
         return conf.discovery_timeout.to(unit);
     }
 
+    public static int getDiscoveryRounds()
+    {
+        return conf.discovery_rounds;
+    }
+
     public static boolean getUnsafeTCMMode()
     {
         return conf.unsafe_tcm_mode;

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

@@ -328,6 +328,11 @@ public boolean isAlive(InetAddressAndPort ep)
             // registration via the metadata log, or a full gossip round). This is perfectly harmless, so no need to log
             // an error in that case.
             ClusterMetadata metadata = ClusterMetadata.current();
+            if (metadata.cmsLookup.isActive() && metadata.fullCMSMembers().contains(ep))
+            {
+                logger.trace("Found endpoint {} in active CMS lookup, assuming it is alive", ep);
+                return true;
+            }
             if (!metadata.directory.allJoinedEndpoints().contains(ep) && !metadata.fullCMSMembers().contains(ep))
                 logger.error("Unknown endpoint: " + ep, new UnknownEndpointException(ep));
         }

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

@@ -56,7 +56,7 @@ public void doVerb(Message<GossipDigestAck> message)
             if (logger.isDebugEnabled())
                 logger.debug("Received an ack from {}, which may trigger exit from shadow round", from);
 
-            NewGossiper.instance.onAck(epStateMap);
+            NewGossiper.instance.onAck(message.from(), epStateMap);
             return;
         }
         if (epStateMap.size() > 0)

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

@@ -720,6 +720,17 @@ public void unsafeAnnulEndpoint(InetAddressAndPort endpoint)
         unreachableEndpoints.remove(endpoint);
     }
 
+    /**
+     * determine which endpoint started up earlier
+     */
+    public int compareEndpointStartup(InetAddressAndPort addr1, InetAddressAndPort addr2)
+    {
+        EndpointState ep1 = getEndpointStateForEndpoint(addr1);
+        EndpointState ep2 = getEndpointStateForEndpoint(addr2);
+        assert ep1 != null && ep2 != null;
+        return ep1.getHeartBeatState().getGeneration() - ep2.getHeartBeatState().getGeneration();
+    }
+
     /**
      * Quarantines the endpoint for QUARANTINE_DELAY
      *
@@ -944,7 +955,7 @@ public boolean isGossipOnlyMember(InetAddressAndPort endpoint)
         ClusterMetadata metadata = ClusterMetadata.current();
         NodeId nodeId = metadata.directory.peerId(endpoint);
         if (nodeId == null)
-            return false;
+            return true;
         return NodeState.isPreJoin(metadata.directory.states.get(nodeId));
     }
 

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

@@ -20,11 +20,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -41,7 +41,6 @@
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.compatibility.GossipHelper;
-import org.apache.cassandra.utils.concurrent.Accumulator;
 import org.apache.cassandra.utils.concurrent.AsyncPromise;
 import org.apache.cassandra.utils.concurrent.Promise;
 
@@ -60,9 +59,14 @@ public class NewGossiper
     public Map<InetAddressAndPort, EndpointState> doShadowRound()
     {
         Set<InetAddressAndPort> peers = new HashSet<>(SystemKeyspace.loadHostIds().keySet());
+        for (InetAddressAndPort seed : DatabaseDescriptor.getSeeds())
+        {
+            if (!seed.equals(getBroadcastAddressAndPort()))
+                peers.add(seed);
+        }
+
+        // implies a single node cluster with only that one node configured as a seed
         if (peers.isEmpty())
-            peers.addAll(DatabaseDescriptor.getSeeds());
-        if (peers.equals(Collections.singleton(getBroadcastAddressAndPort())))
             return GossipHelper.storedEpstate();
 
         ShadowRoundHandler shadowRoundHandler = new ShadowRoundHandler(peers);
@@ -85,6 +89,9 @@ public Map<InetAddressAndPort, EndpointState> doShadowRound()
             }
         }
         logger.warn("Not able to construct initial cluster metadata from gossip, using system tables instead");
+        // Mark done here so that future gossip messages don't get routed to the shadow round handler (see
+        // GossipDigestSynVerbHandler & GossipDigestAckVerbHandler)
+        handler.markDone();
         return GossipHelper.storedEpstate();
     }
 
@@ -94,18 +101,18 @@ public boolean isInShadowRound()
         return srh != null && !srh.isDone();
     }
 
-    void onAck( Map<InetAddressAndPort, EndpointState> epStateMap)
+    void onAck(InetAddressAndPort from, Map<InetAddressAndPort, EndpointState> epStateMap)
     {
         ShadowRoundHandler srh = handler;
         if (srh != null && !srh.isDone())
-            srh.onAck(epStateMap);
+            srh.onAck(from, epStateMap);
     }
 
     public static class ShadowRoundHandler
     {
         private volatile boolean isDone = false;
         private final Set<InetAddressAndPort> peers;
-        private final Accumulator<Map<InetAddressAndPort, EndpointState>> responses;
+        private final Map<InetAddressAndPort, Map<InetAddressAndPort, EndpointState>> responses;
         private final int requiredResponses;
         private final MessageDelivery messageDelivery;
         private final Promise<Map<InetAddressAndPort, EndpointState>> promise = new AsyncPromise<>();
@@ -117,12 +124,19 @@ public ShadowRoundHandler(Set<InetAddressAndPort> peers)
 
         public ShadowRoundHandler(Set<InetAddressAndPort> peers, MessageDelivery messageDelivery)
         {
-            this.peers = peers;
-            requiredResponses = Math.max(peers.size() / 10, 1); // todo: is 10% reasonable?
-            responses = new Accumulator<>(requiredResponses);
+            this.peers = ConcurrentHashMap.newKeySet();
+            this.peers.addAll(peers);
+            responses = new ConcurrentHashMap<>();
+            requiredResponses = this.peers.size() <= 3 ? 1 : Math.max(this.peers.size() / 5, 2); // require response from 20% of the cluster
             this.messageDelivery = messageDelivery;
         }
 
+        public void markDone()
+        {
+            logger.info("Marking NewGossiper shadow round done");
+            isDone = true;
+        }
+
         public boolean isDone()
         {
             return isDone;
@@ -146,18 +160,21 @@ public Promise<Map<InetAddressAndPort, EndpointState>> doShadowRound()
             return promise;
         }
 
-        public void onAck(Map<InetAddressAndPort, EndpointState> epStateMap)
+        public void onAck(InetAddressAndPort from, Map<InetAddressAndPort, EndpointState> epStateMap)
         {
             if (!isDone)
             {
                 if (!epStateMap.isEmpty())
-                    responses.add(epStateMap);
+                {
+                    responses.put(from, epStateMap);
+                    peers.addAll(epStateMap.keySet()); // when retrying we should query the endpoints we learned about in the previous round
+                }
 
                 logger.debug("Received {} responses. {} required.", responses.size(), requiredResponses);
                 if (responses.size() >= requiredResponses)
                 {
                     isDone = true;
-                    Map<InetAddressAndPort, EndpointState> merged = merge(responses.snapshot());
+                    Map<InetAddressAndPort, EndpointState> merged = merge(responses.values());
                     if (GossipHelper.isValidForClusterMetadata(merged))
                         promise.setSuccess(merged);
                     else

src/java/org/apache/cassandra/net/MessageDelivery.java

@@ -53,6 +53,7 @@ static <REQ, RSP> Collection<Pair<InetAddressAndPort, RSP>> fanoutAndWait(Messag
     {
         return fanoutAndWait(messaging, sendTo, verb, payload, DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
     }
+
     static <REQ, RSP> Collection<Pair<InetAddressAndPort, RSP>> fanoutAndWait(MessageDelivery messaging, Set<InetAddressAndPort> sendTo, Verb verb, REQ payload, long timeout, TimeUnit timeUnit)
     {
         Accumulator<Pair<InetAddressAndPort, RSP>> responses = new Accumulator<>(sendTo.size());
@@ -62,21 +63,21 @@ static <REQ, RSP> Collection<Pair<InetAddressAndPort, RSP>> fanoutAndWait(Messag
             @Override
             public void onResponse(Message<RSP> msg)
             {
-                logger.info("Received a {} response from {}: {}", msg.verb(), msg.from(), msg.payload);
+                logger.debug("Received a {} response from {}: {}", msg.verb(), msg.from(), msg.payload);
                 responses.add(Pair.create(msg.from(), msg.payload));
                 cdl.decrement();
             }
 
             @Override
             public void onFailure(InetAddressAndPort from, RequestFailure reason)
             {
-                logger.info("Received failure in response to {} from {}: {}", verb, from, reason);
+                logger.debug("Received failure in response to {} from {}: {}", verb, from, reason);
                 cdl.decrement();
             }
         };
 
         sendTo.forEach((ep) -> {
-            logger.info("Election for metadata migration sending {} ({}) to {}", verb, payload.toString(), ep);
+            logger.debug("Sending {} ({}) to {}", verb, payload.toString(), ep);
             messaging.sendWithCallback(Message.out(verb, payload), ep, callback);
         });
         cdl.awaitUninterruptibly(timeout, timeUnit);

src/java/org/apache/cassandra/net/ResponseVerbHandler.java

@@ -48,6 +48,8 @@ class ResponseVerbHandler implements IVerbHandler
                                                                  Verb.TCM_REPLICATION,
                                                                  Verb.TCM_NOTIFY_RSP,
                                                                  Verb.TCM_DISCOVER_RSP,
+                                                                 Verb.TCM_DISCOVER_PEERS_RSP,
+                                                                 Verb.TCM_DISCOVER_SURVEY_RSP,
                                                                  Verb.TCM_INIT_MIG_RSP);
 
     // We skip epoch catchup for PaxosV2 verbs, since we are using PaxosV2 to serially read the log.

src/java/org/apache/cassandra/net/Verb.java

@@ -136,10 +136,13 @@
 import org.apache.cassandra.streaming.DataMovement;
 import org.apache.cassandra.streaming.DataMovementVerbHandler;
 import org.apache.cassandra.streaming.ReplicationDoneVerbHandler;
-import org.apache.cassandra.tcm.Discovery;
 import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.FetchCMSLog;
 import org.apache.cassandra.tcm.FetchPeerLog;
+import org.apache.cassandra.tcm.discovery.Discovery;
+import org.apache.cassandra.tcm.discovery.SurveyRequest;
+import org.apache.cassandra.tcm.discovery.SurveyRequestHandler;
+import org.apache.cassandra.tcm.discovery.SurveyResponse;
 import org.apache.cassandra.tcm.migration.CMSInitializationRequest;
 import org.apache.cassandra.tcm.migration.CMSInitializationResponse;
 import org.apache.cassandra.tcm.migration.Election;
@@ -316,6 +319,10 @@ public enum Verb
     TCM_DISCOVER_REQ       (813, P0, rpcTimeout,      INTERNAL_METADATA,    () -> NoPayload.serializer,                         () -> Discovery.instance.requestHandler,    TCM_DISCOVER_RSP       ),
     TCM_FETCH_PEER_LOG_RSP (818, P0, shortTimeout,    FETCH_METADATA,       MessageSerializers::logStateSerializer,             RESPONSE_HANDLER                                 ),
     TCM_FETCH_PEER_LOG_REQ (819, P0, rpcTimeout,      FETCH_METADATA,       () -> FetchPeerLog.serializer,                      () -> FetchPeerLog.Handler.instance,        TCM_FETCH_PEER_LOG_RSP ),
+    TCM_DISCOVER_PEERS_RSP (820, P0, rpcTimeout,      INTERNAL_METADATA,    () -> Discovery.serializer,                         () -> ResponseVerbHandler.instance                                 ),
+    TCM_DISCOVER_PEERS_REQ (821, P0, rpcTimeout,      INTERNAL_METADATA,    () -> NoPayload.serializer,                         () -> Discovery.instance.requestHandler,    TCM_DISCOVER_PEERS_RSP),
+    TCM_DISCOVER_SURVEY_RSP(822, P0, rpcTimeout,      INTERNAL_METADATA,    () -> SurveyResponse.serializer,                    () -> ResponseVerbHandler.instance                                 ),
+    TCM_DISCOVER_SURVEY_REQ(823, P0, rpcTimeout,      INTERNAL_METADATA,    () -> SurveyRequest.serializer,                     () -> SurveyRequestHandler.instance(),      TCM_DISCOVER_SURVEY_RSP),
 
     INITIATE_DATA_MOVEMENTS_RSP (814, P1, rpcTimeout, MISC, () -> NoPayload.serializer,             RESPONSE_HANDLER                                  ),
     INITIATE_DATA_MOVEMENTS_REQ (815, P1, rpcTimeout, MISC, () -> DataMovement.serializer,          () -> DataMovementVerbHandler.instance, INITIATE_DATA_MOVEMENTS_RSP ),