[CASSANDRA-20476] Attempt to wait for all address changes to be enacted before allowing startup to proceed

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

@@ -328,6 +328,11 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean
             // 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/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/tcm/Startup.java

@@ -122,17 +122,47 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
                 break;
             case NORMAL:
                 logger.info("Initializing as non CMS node");
+                // note: if this node was a member of the CMS, log replay will put it back into that state
                 initializeAsNonCmsNode(wrapProcessor);
-                ClusterMetadataService.instance().log().pause();
-                initMessaging.run();
                 UUID localHostId = SystemKeyspace.getLocalHostId();
                 // If null, this node has not yet registered so there will be more nothing to do here
                 if (localHostId != null)
                 {
                     NodeId nodeId = NodeId.fromUUID(localHostId);
-                    initializeCMSLookup(nodeId, ClusterMetadata.current());
+                    ClusterMetadata replayed = ClusterMetadata.current();
+                    InetAddressAndPort oldAddress = replayed.directory.endpoint(nodeId);
+                    InetAddressAndPort newAddress = FBUtilities.getBroadcastAddressAndPort();
+                    if (!newAddress.equals(oldAddress))
+                    {
+                        // Build temporary mappings for addressing CMS nodes who's addresses have been
+                        // changed but not yet committed via the CMS or not yet been enacted locally.
+                        ClusterMetadataService.instance().log().pause();
+                        initMessaging.run();
+                        initializeCMSLookup(nodeId, replayed);
+                        ClusterMetadataService.instance().log().unpause();
+
+                        logger.info("Detected change in local node addresses, committing update to Cluster Metadata Service");
+                        Transformation transform = new org.apache.cassandra.tcm.transformations.Startup(nodeId,
+                                                                                                        NodeAddresses.current(),
+                                                                                                        NodeVersion.CURRENT);
+                        replayed = ClusterMetadataService.instance().commit(transform);
+
+                        // Wait for any CMS members which have uncommitted address changes to commit them and
+                        // us to enact them.
+                        while (replayed.cmsLookup.isActive())
+                        {
+                            logger.info("Waiting for pending CMS address changes to complete {}", replayed.cmsLookup);
+                            TimeUnit.MILLISECONDS.sleep(1000);  // TODO make configurable?
+                            replayed = ClusterMetadata.current();
+                        }
+                    }
+                    logger.info("CMS address changes complete, current epoch is {}", replayed.epoch.getEpoch());
+                }
+                else
+                {
+                    // nothing to do, so just initialise messaging
+                    initMessaging.run();
                 }
-                ClusterMetadataService.instance().log().unpause();
                 break;
             case VOTE:
                 logger.info("Initializing for discovery");
@@ -282,6 +312,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
             surveyed.forEach(pair -> {
                 if (previousCMS.containsKey(pair.right))
                     confirmedCMS.put(pair.right, pair.left);
+
             });
 
             logger.info("Confirmed CMS members {}", confirmedCMS);

test/distributed/org/apache/cassandra/distributed/test/log/DiscoverNewCMSTest.java

@@ -103,12 +103,20 @@ public class DiscoverNewCMSTest extends TestBaseImpl
     private void test(Cluster cluster, int cmsSize) throws IOException, ExecutionException, InterruptedException
     {
         ExecutorService executor = Executors.newFixedThreadPool(cluster.size());
+        // Some fetchCMSLog operations might temporarily fail and be retried during address changes
+        String[] expectedErrors = new String[] {
+            "Cannot achieve consistency level SERIAL.*",
+            "Queried for epoch Epoch\\{epoch=\\d+\\}, but could not catch up.*"
+        };
         cluster.setUncaughtExceptionsFilter((node, t) -> {
             Throwable rootCause = Throwables.getRootCause(t);
-            // Some fetchCMSLog operations might temporarily fail and be retried during address changes
+            if (rootCause.getMessage() == null)
-            return rootCause.getMessage() != null
+                return false;
-                   && rootCause.getMessage().startsWith("Cannot achieve consistency level SERIAL");
+            String message = rootCause.getMessage();
-
+            for (String expected : expectedErrors)
+                if (message.matches(expected))
+                    return true;
+            return false;
         });
         cluster.startup();
         init(cluster);