From b791644fda91b343a679bb0c2c1e33e594524636 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Thu, 4 May 2023 05:46:11 -0500 Subject: [PATCH] Allow internal address to change with reconnecting snitches Patch by brandonwilliams; reviewed by bereng for CASSANDRA-16718 --- CHANGES.txt | 1 + .../org/apache/cassandra/gms/EndpointState.java | 5 +++++ src/java/org/apache/cassandra/gms/Gossiper.java | 17 +++++++++++++++++ .../locator/ReconnectableSnitchHelper.java | 8 ++++++-- .../apache/cassandra/net/MessagingService.java | 7 +++++++ .../net/OutboundConnectionSettings.java | 4 ++-- 6 files changed, 38 insertions(+), 4 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index f6c5cfc757..ff745f3092 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.10 + * Allow internal address to change with reconnecting snitches (CASSANDRA-16718) * Fix quoting in toCqlString methods of UDTs and aggregates (CASSANDRA-17918) * NPE when deserializing malformed collections from client (CASSANDRA-18505) * Improve 'Not enough space for compaction' logging messages (CASSANDRA-18260) diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java index b8d56263e7..fe29c4bbcc 100644 --- a/src/java/org/apache/cassandra/gms/EndpointState.java +++ b/src/java/org/apache/cassandra/gms/EndpointState.java @@ -86,6 +86,11 @@ public class EndpointState return applicationState.get().get(key); } + public void removeApplicationState(ApplicationState key) + { + applicationState.get().remove(key); + } + public boolean containsApplicationState(ApplicationState key) { return applicationState.get().containsKey(key); diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index cf505bb46c..ed5788e6ba 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -1154,6 +1154,23 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return UUID.fromString(epStates.get(endpoint).getApplicationState(ApplicationState.HOST_ID).value); } + public InetAddressAndPort getInternalAddressAndPort(InetAddressAndPort endpoint) + { + try + { + String internal = getApplicationState(endpoint, ApplicationState.INTERNAL_ADDRESS_AND_PORT); + if (internal == null) + internal = getApplicationState(endpoint, ApplicationState.INTERNAL_IP); + if (internal == null) + return endpoint; + return InetAddressAndPort.getByName(internal); + } + catch (UnknownHostException e) + { + throw new RuntimeException(e); + } + } + /** * The value for the provided application state for the provided endpoint as currently known by this Gossip instance. * diff --git a/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java index dea8c76f4e..7bb38cb491 100644 --- a/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java +++ b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java @@ -91,7 +91,7 @@ public class ReconnectableSnitchHelper implements IEndpointStateChangeSubscriber VersionedValue address = epState.getApplicationState(ApplicationState.INTERNAL_ADDRESS_AND_PORT); if (address == null) { - address = epState.getApplicationState(ApplicationState.INTERNAL_ADDRESS_AND_PORT); + address = epState.getApplicationState(ApplicationState.INTERNAL_IP); } if (address != null) { @@ -129,7 +129,11 @@ public class ReconnectableSnitchHelper implements IEndpointStateChangeSubscriber public void onDead(InetAddressAndPort endpoint, EndpointState state) { - // do nothing. + if (preferLocal && state.getApplicationState(ApplicationState.INTERNAL_ADDRESS_AND_PORT) != null) + state.removeApplicationState(ApplicationState.INTERNAL_ADDRESS_AND_PORT); + if (preferLocal && state.getApplicationState(ApplicationState.INTERNAL_IP) != null) + state.removeApplicationState(ApplicationState.INTERNAL_IP); + MessagingService.instance().closeOutboundNow(endpoint); } public void onRemove(InetAddressAndPort endpoint) diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 71f2231eb2..cd124ad719 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -370,6 +370,13 @@ public final class MessagingService extends MessagingServiceMBeanImpl .addListener(future -> channelManagers.remove(to, pool)); } + public void closeOutboundNow(InetAddressAndPort to) + { + OutboundConnections pool = channelManagers.get(to); + if (pool != null) + closeOutboundNow(pool); + } + /** * Only to be invoked once we believe the connections will never be used again. */ diff --git a/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java b/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java index 92a1f52bc3..d9038ee487 100644 --- a/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java +++ b/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java @@ -27,7 +27,7 @@ import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.EncryptionOptions; import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions; -import org.apache.cassandra.db.SystemKeyspace; +import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.locator.IEndpointSnitch; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.utils.FBUtilities; @@ -452,7 +452,7 @@ public class OutboundConnectionSettings { InetAddressAndPort connectTo = this.connectTo; if (connectTo == null) - connectTo = SystemKeyspace.getPreferredIP(to); + connectTo = Gossiper.instance.getInternalAddressAndPort(to); if (FBUtilities.getBroadcastAddressAndPort().equals(connectTo)) return FBUtilities.getLocalAddressAndPort(); return connectTo;