natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow internal address to change with reconnecting snitches 

Patch by brandonwilliams; reviewed by bereng for CASSANDRA-16718
This commit is contained in:
Brandon Williams 2023-05-04 05:46:11 -05:00
parent 2d40ee9019
commit b791644fda
6 changed files with 38 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

5
src/java/org/apache/cassandra/gms/EndpointState.java
View File

@ -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);

17
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -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.
*

8
src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java
View File

@ -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)

7
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -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.
*/

4
src/java/org/apache/cassandra/net/OutboundConnectionSettings.java
View File

@ -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;