You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "beobal (via GitHub)" <gi...@apache.org> on 2026/07/30 10:05:04 UTC
[PR] CASSANDRA-20476 Cluster is unable to recover after shutdown if IPs change [cassandra]
beobal opened a new pull request, #4979:
URL: https://github.com/apache/cassandra/pull/4979
Thanks for sending a pull request! Here are some tips if you're new here:
* Ensure you have added or run the [appropriate tests](https://cassandra.apache.org/_/development/testing.html) for your PR.
* Be sure to keep the PR description updated to reflect all changes.
* Write your PR title to summarize what this PR proposes.
* If possible, provide a concise example to reproduce the issue for a faster review.
* Read our [contributor guidelines](https://cassandra.apache.org/_/development/index.html)
* If you're making a documentation change, see our [guide to documentation contribution](https://cassandra.apache.org/_/development/documentation.html)
Commit messages should follow the following format:
```
<One sentence description, usually Jira title or CHANGES.txt summary>
<Optional lengthier description (context on patch)>
patch by <Authors>; reviewed by <Reviewers> for CASSANDRA-#####
Co-authored-by: Name1 <email1>
Co-authored-by: Name2 <email2>
```
The [Cassandra Jira](https://issues.apache.org/jira/projects/CASSANDRA/issues/)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-20476 Cluster is unable to recover after shutdown if IPs change [cassandra]
Posted by "krummas (via GitHub)" <gi...@apache.org>.
krummas commented on code in PR #4979:
URL: https://github.com/apache/cassandra/pull/4979#discussion_r3703987986
##########
src/java/org/apache/cassandra/tcm/CMSLookup.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.tcm.listeners.ChangeListener;
+import org.apache.cassandra.tcm.membership.EndpointLookup;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.Pair;
+
+public class CMSLookup
+{
+ private static final Logger logger = LoggerFactory.getLogger(CMSLookup.class);
+
+ public enum State { PRE_INIT, ACTIVE, RETIRED };
+
+ public final static CMSLookup NO_OP = new CMSLookup(State.PRE_INIT, Epoch.EMPTY, ImmutableMap.of());
+ public static InitialBuilder builder(ClusterMetadata metadata)
+ {
+ return new InitialBuilder(metadata);
+ }
+
+ private final ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> overrides;
+ private final Epoch lastModified;
+ private final State state;
+
+ private CMSLookup(State state, Epoch epoch, ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> overrides)
+ {
+ this.state = state;
+ this.lastModified = epoch;
+ this.overrides = overrides;
+ }
+
+ public boolean isUninitialized()
+ {
+ return state == State.PRE_INIT;
+ }
+
+ public boolean isActive()
+ {
+ return state == State.ACTIVE;
+ }
+
+ public EndpointLookup asNodeLookup(EndpointLookup lookup)
+ {
+ return new EndpointLookup()
+ {
+ @Override
+ public InetAddressAndPort endpoint(NodeId id)
+ {
+ if (overrides.containsKey(id))
+ return overrides.get(id).right;
+ return lookup.endpoint(id);
+ }
+ };
+ }
+
+ public CMSLookup rebuild(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
+ {
+ logger.info("Rebuilding CMS lookup {} with metadata from epoch {}", this, next.epoch.getEpoch());
+
+ // All address changes have been enacted, nothing to do
+ if (state == State.RETIRED)
+ return this;
+
+ if (!next.epoch.isEqualOrBefore(Epoch.FIRST)
+ && !fromSnapshot
+ && next.directory.lastModified().equals(prev.directory.lastModified()))
+ return this;
+
+ // Filters from the override list those which are no longer necessary as a transformation has now
+ // replaced the old address with a new one for that node id
+ Predicate<Map.Entry<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>>> overrideRequired = entry -> {
+ NodeId nodeId = entry.getKey();
+ if (!Objects.equals(prev.directory.getNodeAddresses(nodeId), next.directory.getNodeAddresses(nodeId)))
+ {
+ Pair<InetAddressAndPort, InetAddressAndPort> override = overrides.get(nodeId);
+ if (override != null)
+ {
+ // If there was an override for this nodeId && the address being overriden matches the prev
+ // directory entry, filter out the override from the map which will be used to build the new version
+ // (i.e. return false from Predicate::test). This indicates the override is no longer required.
+ return !override.left.equals(prev.directory.endpoint(nodeId));
+ }
+ }
+ return true;
+ };
+
+ logger.info("Current endpoint overrides: {}", overrides);
+ ImmutableMap.Builder<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> builder = ImmutableMap.builder();
+ for (Map.Entry<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> override : overrides.entrySet())
+ {
+ if(overrideRequired.test(override))
+ builder.put(override.getKey(), override.getValue());
+ }
+
+ ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> nextOverrides = builder.build();
+ if (nextOverrides.equals(overrides))
+ {
+ logger.info("No changes to endpoint overrides detected");
+ return this;
+ }
+
+ logger.info("Proposed endpoint overrides: {}", nextOverrides);
+ State state = nextOverrides.isEmpty() ? State.RETIRED : State.ACTIVE;
Review Comment:
nope, a node that is gone forever will never make it in to the overrides
##########
src/java/org/apache/cassandra/net/Verb.java:
##########
@@ -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 ),
Review Comment:
looks like this should now be `RESPONSE_HANDLER`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-20476 Cluster is unable to recover after shutdown if IPs change [cassandra]
Posted by "beobal (via GitHub)" <gi...@apache.org>.
beobal closed pull request #4979: CASSANDRA-20476 Cluster is unable to recover after shutdown if IPs change
URL: https://github.com/apache/cassandra/pull/4979
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-20476 Cluster is unable to recover after shutdown if IPs change [cassandra]
Posted by "krummas (via GitHub)" <gi...@apache.org>.
krummas commented on code in PR #4979:
URL: https://github.com/apache/cassandra/pull/4979#discussion_r3690592046
##########
src/java/org/apache/cassandra/tcm/RemoteProcessor.java:
##########
@@ -430,6 +434,7 @@ public CandidateIterator(Collection<InetAddressAndPort> initialContacts)
public CandidateIterator(Collection<InetAddressAndPort> initialContacts, boolean checkLive)
{
this.candidates = new ConcurrentLinkedDeque<>(initialContacts);
+ this.elements = new HashSet<>(initialContacts);
Review Comment:
I think this needs to be a concurrent set
##########
src/java/org/apache/cassandra/gms/NewGossiper.java:
##########
@@ -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;
Review Comment:
`markDone()` instead to get the logging?
##########
src/java/org/apache/cassandra/tcm/discovery/SurveyRequestHandler.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.discovery;
+
+import java.io.IOException;
+import java.util.function.Supplier;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.net.IVerbHandler;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessageDelivery;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
+
+public class SurveyRequestHandler implements IVerbHandler<SurveyRequest>
+{
+ private static final Logger logger = LoggerFactory.getLogger(SurveyRequestHandler.class);
+ private static volatile SurveyRequestHandler instance;
+
+ final Supplier<MessageDelivery> messaging;
+ public final int metadataId;
+
+ public static SurveyRequestHandler instance()
+ {
+ if (instance == null)
+ {
+ synchronized (SurveyRequestHandler.class)
+ {
+ if (instance == null)
+ instance = new SurveyRequestHandler();
+ }
+ }
+ return instance;
+ }
+
+ private SurveyRequestHandler()
+ {
+ this(ClusterMetadata.current().metadataIdentifier, MessagingService::instance);
+ }
+
+ public SurveyRequestHandler(int metadataId, Supplier<MessageDelivery> messaging)
Review Comment:
nit, @ VisibleForTesting
##########
src/java/org/apache/cassandra/tcm/ClusterMetadata.java:
##########
@@ -239,7 +245,25 @@ public Set<NodeId> fullCMSMemberIds()
public boolean isCMSMember()
{
- if (epoch.isEqualOrBefore(Epoch.FIRST))
+ // There are two special cases to consider here:
+ // * Initialization of the CMS for the first time by its first member
+ // During first initialization, placements are hardcoded to the local address after the PRE_INITIALIZE_CMS is
+ // committed so that the subsequent INITIALIZE_CMS can be.
+ //
+ // * Some changes to the addresses of the CMS members have been changed and whilst this is known (i.e. has been
Review Comment:
nit, `Some addresses of the CMS members have been changed...`
##########
src/java/org/apache/cassandra/tcm/ClusterMetadata.java:
##########
@@ -275,15 +300,54 @@ public EndpointsForRange fullCMSMembersAsReplicas()
if (fullCMSReplicas == null)
{
+ EndpointLookup lookup = endpointLookup();
EndpointsForRange.Builder builder = EndpointsForRange.builder(MetaStrategy.entireRange);
for (NodeId nodeId : fullCMSMemberIds())
- builder.add(MetaStrategy.replica(directory.endpoint(nodeId)));
+ builder.add(MetaStrategy.replica(lookup.endpoint(nodeId)));
fullCMSReplicas = builder.build();
}
return fullCMSReplicas;
}
- private DataPlacement calculateCMSPlacement(DataPlacements placements, CMSMembership cms)
+ // Synchronization is probably not necessary as this should only be called by a log listener
+ public synchronized void refreshCMSLookup(ClusterMetadata prev, boolean fromSnapshot)
+ {
+ CMSLookup prevLookup = prev.cmsLookup;
+ CMSLookup proposedLookup = prevLookup.rebuild(prev, this, fromSnapshot);
+ // rebuild returns `this` if no changes
+ if (prevLookup != proposedLookup)
+ logger.info("Replacing CMSLookup, Current: {}, Proposed: {}", prevLookup, proposedLookup);
+
+ // recalculate CMS placement using endpoint mappings from lookup
+ cmsDataPlacement = calculateCMSPlacement(placements, cmsMembership, proposedLookup);
+
+ // We shouldn't need to null out the other lazily initialized CMS fields when we refresh the CMSLookup as the
Review Comment:
should we assert that they are null?
##########
src/java/org/apache/cassandra/tcm/Commit.java:
##########
@@ -449,21 +452,19 @@ public void send(Result result, InetAddressAndPort source)
Result.Success success = result.success();
Directory directory = directorySupplier.get();
+ EndpointLookup lookup = lookupSupplier.get();
Review Comment:
is there a possibility that we get "mismatching" `directory`/`lookup` here since we call suppliers twice? Maybe the supplier should be something like `Supplier<Pair<Directory, EndpointLookup>>` to make sure it's backed by the same ClusterMetadata?
##########
src/java/org/apache/cassandra/tcm/CMSLookup.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.tcm.listeners.ChangeListener;
+import org.apache.cassandra.tcm.membership.EndpointLookup;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.Pair;
+
+public class CMSLookup
+{
+ private static final Logger logger = LoggerFactory.getLogger(CMSLookup.class);
+
+ public enum State { PRE_INIT, ACTIVE, RETIRED };
+
+ public final static CMSLookup NO_OP = new CMSLookup(State.PRE_INIT, Epoch.EMPTY, ImmutableMap.of());
+ public static InitialBuilder builder(ClusterMetadata metadata)
+ {
+ return new InitialBuilder(metadata);
+ }
+
+ private final ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> overrides;
+ private final Epoch lastModified;
+ private final State state;
+
+ private CMSLookup(State state, Epoch epoch, ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> overrides)
+ {
+ this.state = state;
+ this.lastModified = epoch;
+ this.overrides = overrides;
+ }
+
+ public boolean isUninitialized()
+ {
+ return state == State.PRE_INIT;
+ }
+
+ public boolean isActive()
+ {
+ return state == State.ACTIVE;
+ }
+
+ public EndpointLookup asNodeLookup(EndpointLookup lookup)
+ {
+ return new EndpointLookup()
+ {
+ @Override
+ public InetAddressAndPort endpoint(NodeId id)
+ {
+ if (overrides.containsKey(id))
Review Comment:
nit, avoid looking up the key twice:
```
Pair<InetAddressAndPort, InetAddressAndPort> override = overrides.get(id);
if (override != null)
return override.right;
return lookup.endpoint(id);
...
##########
src/java/org/apache/cassandra/tcm/Startup.java:
##########
@@ -117,8 +125,52 @@ public static void initialize(Set<InetAddressAndPort> seeds,
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);
- 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)
+ {
+ initMessaging.run();
+ }
+ else
+ {
+ NodeId nodeId = NodeId.fromUUID(localHostId);
+ ClusterMetadata replayed = ClusterMetadata.current();
+ InetAddressAndPort oldAddress = replayed.directory.endpoint(nodeId);
+ InetAddressAndPort newAddress = FBUtilities.getBroadcastAddressAndPort();
+ if (!newAddress.equals(oldAddress) && replayed.directory.commonSerializationVersion.isAtLeast(Version.V9))
+ {
+ // 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);
+ replayed = ClusterMetadata.current();
Review Comment:
Is there a chance we miss a `Startup` and get stuck forever here? Should we do an explicit log fetch periodically? Maybe not every second, but once a minute or so, just to make sure?
##########
src/java/org/apache/cassandra/tcm/CMSLookup.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.tcm.listeners.ChangeListener;
+import org.apache.cassandra.tcm.membership.EndpointLookup;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.Pair;
+
+public class CMSLookup
+{
+ private static final Logger logger = LoggerFactory.getLogger(CMSLookup.class);
+
+ public enum State { PRE_INIT, ACTIVE, RETIRED };
+
+ public final static CMSLookup NO_OP = new CMSLookup(State.PRE_INIT, Epoch.EMPTY, ImmutableMap.of());
+ public static InitialBuilder builder(ClusterMetadata metadata)
+ {
+ return new InitialBuilder(metadata);
+ }
+
+ private final ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> overrides;
+ private final Epoch lastModified;
+ private final State state;
+
+ private CMSLookup(State state, Epoch epoch, ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> overrides)
+ {
+ this.state = state;
+ this.lastModified = epoch;
+ this.overrides = overrides;
+ }
+
+ public boolean isUninitialized()
+ {
+ return state == State.PRE_INIT;
+ }
+
+ public boolean isActive()
+ {
+ return state == State.ACTIVE;
+ }
+
+ public EndpointLookup asNodeLookup(EndpointLookup lookup)
+ {
+ return new EndpointLookup()
+ {
+ @Override
+ public InetAddressAndPort endpoint(NodeId id)
+ {
+ if (overrides.containsKey(id))
+ return overrides.get(id).right;
+ return lookup.endpoint(id);
+ }
+ };
+ }
+
+ public CMSLookup rebuild(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
+ {
+ logger.info("Rebuilding CMS lookup {} with metadata from epoch {}", this, next.epoch.getEpoch());
+
+ // All address changes have been enacted, nothing to do
+ if (state == State.RETIRED)
+ return this;
+
+ if (!next.epoch.isEqualOrBefore(Epoch.FIRST)
+ && !fromSnapshot
+ && next.directory.lastModified().equals(prev.directory.lastModified()))
+ return this;
+
+ // Filters from the override list those which are no longer necessary as a transformation has now
+ // replaced the old address with a new one for that node id
+ Predicate<Map.Entry<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>>> overrideRequired = entry -> {
+ NodeId nodeId = entry.getKey();
+ if (!Objects.equals(prev.directory.getNodeAddresses(nodeId), next.directory.getNodeAddresses(nodeId)))
+ {
+ Pair<InetAddressAndPort, InetAddressAndPort> override = overrides.get(nodeId);
+ if (override != null)
+ {
+ // If there was an override for this nodeId && the address being overriden matches the prev
+ // directory entry, filter out the override from the map which will be used to build the new version
+ // (i.e. return false from Predicate::test). This indicates the override is no longer required.
+ return !override.left.equals(prev.directory.endpoint(nodeId));
+ }
+ }
+ return true;
+ };
+
+ logger.info("Current endpoint overrides: {}", overrides);
+ ImmutableMap.Builder<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> builder = ImmutableMap.builder();
+ for (Map.Entry<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> override : overrides.entrySet())
+ {
+ if(overrideRequired.test(override))
+ builder.put(override.getKey(), override.getValue());
+ }
+
+ ImmutableMap<NodeId, Pair<InetAddressAndPort, InetAddressAndPort>> nextOverrides = builder.build();
+ if (nextOverrides.equals(overrides))
+ {
+ logger.info("No changes to endpoint overrides detected");
+ return this;
+ }
+
+ logger.info("Proposed endpoint overrides: {}", nextOverrides);
+ State state = nextOverrides.isEmpty() ? State.RETIRED : State.ACTIVE;
Review Comment:
Could/should we retire the lookup once we have a quorum committed? This would allow us to recover if one of the CMS nodes is gone forever
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-20476 Cluster is unable to recover after shutdown if IPs change [cassandra]
Posted by "beobal (via GitHub)" <gi...@apache.org>.
beobal commented on code in PR #4979:
URL: https://github.com/apache/cassandra/pull/4979#discussion_r3704816235
##########
src/java/org/apache/cassandra/tcm/Startup.java:
##########
@@ -117,8 +125,52 @@ public static void initialize(Set<InetAddressAndPort> seeds,
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);
- 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)
+ {
+ initMessaging.run();
+ }
+ else
+ {
+ NodeId nodeId = NodeId.fromUUID(localHostId);
+ ClusterMetadata replayed = ClusterMetadata.current();
+ InetAddressAndPort oldAddress = replayed.directory.endpoint(nodeId);
+ InetAddressAndPort newAddress = FBUtilities.getBroadcastAddressAndPort();
+ if (!newAddress.equals(oldAddress) && replayed.directory.commonSerializationVersion.isAtLeast(Version.V9))
+ {
+ // 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);
+ replayed = ClusterMetadata.current();
Review Comment:
I think there is a slight chance of that happening, but I don't think we need to wait for the commit/enactment anyway. By this point we must have already successfully committed our own address change, so we can just exit and allow any other inflight changes to be committed/propagated/enacted in the background.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org