diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index 13d6266ca0..34f5350809 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/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)); } diff --git a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java index 215b575c10..d44c9510f1 100644 --- a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java +++ b/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. diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java index b87cca6614..f52d6c6bb2 100644 --- a/src/java/org/apache/cassandra/tcm/Startup.java +++ b/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); diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/DiscoverNewCMSTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/DiscoverNewCMSTest.java index f73465aa47..3474ea9838 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/DiscoverNewCMSTest.java +++ b/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 - return rootCause.getMessage() != null - && rootCause.getMessage().startsWith("Cannot achieve consistency level SERIAL"); - + if (rootCause.getMessage() == null) + return false; + String message = rootCause.getMessage(); + for (String expected : expectedErrors) + if (message.matches(expected)) + return true; + return false; }); cluster.startup(); init(cluster);