Allow GPFS to reconnect to the internal IP.

Patch by Chris Burroughs, reviewed by brandonwilliams for CASSANDRA-5630

CHANGES.txt

@@ -22,6 +22,7 @@
  * Evaluate now() function at execution time (CASSANDRA-5616)
  * Expose detailed read repair metrics (CASSANDRA-5618)
  * Correct blob literal + ReversedType parsing (CASSANDRA-5629)
+ * Allow GPFS to prefer the internal IP like EC2MRS (CASSANDRA-5630)
 Merged from 1.1:
  * Remove buggy thrift max message length option (CASSANDRA-5529)
  * Fix NPE in Pig's widerow mode (CASSANDRA-5488)

conf/cassandra-rackdc.properties

@@ -22,3 +22,6 @@ rack=RAC1
 # Add a suffix to a datacenter name. Used by the Ec2Snitch and Ec2MultiRegionSnitch
 # to append a string to the EC2 region name.
 #dc_suffix=
+
+# Uncomment the following line to make this snitch prefer the internal ip when possible, as the Ec2MultiRegionSnitch does.
+# prefer_local=true

src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java

@@ -19,6 +19,7 @@
 package org.apache.cassandra.locator;
 
 import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.util.Map;
 
 import org.apache.cassandra.db.SystemTable;
@@ -29,9 +30,14 @@ import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.gms.ApplicationState;
 import org.apache.cassandra.gms.EndpointState;
 import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
+import org.apache.cassandra.gms.VersionedValue;
+import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.service.StorageService;
 
-public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch
+
+public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch implements IEndpointStateChangeSubscriber
 {
     private static final Logger logger = LoggerFactory.getLogger(GossipingPropertyFileSnitch.class);
 
@@ -41,6 +47,7 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch
     private Map<InetAddress, Map<String, String>> savedEndpoints;
     private String DEFAULT_DC = "UNKNOWN_DC";
     private String DEFAULT_RACK = "UNKNOWN_RACK";
+    private boolean preferLocal;
 
     public GossipingPropertyFileSnitch() throws ConfigurationException
     {
@@ -51,7 +58,7 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch
 
         myDC = myDC.trim();
         myRack = myRack.trim();
-
+        preferLocal = Boolean.parseBoolean(SnitchProperties.get("prefer_local", "false"));
         try
         {
             psnitch = new PropertyFileSnitch();
@@ -118,4 +125,65 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch
         }
         return epState.getApplicationState(ApplicationState.RACK).value;
     }
+
+    // IEndpointStateChangeSubscriber methods
+
+    public void onJoin(InetAddress endpoint, EndpointState epState)
+    {
+        if (preferLocal && epState.getApplicationState(ApplicationState.INTERNAL_IP) != null)
+            reConnect(endpoint, epState.getApplicationState(ApplicationState.INTERNAL_IP));
+    }
+
+    public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value)
+    {
+        if (preferLocal && state == ApplicationState.INTERNAL_IP)
+            reConnect(endpoint, value);
+    }
+
+    public void onAlive(InetAddress endpoint, EndpointState state)
+    {
+        if (preferLocal && state.getApplicationState(ApplicationState.INTERNAL_IP) != null)
+            reConnect(endpoint, state.getApplicationState(ApplicationState.INTERNAL_IP));
+    }
+
+    public void onDead(InetAddress endpoint, EndpointState state)
+    {
+        // do nothing
+    }
+
+    public void onRestart(InetAddress endpoint, EndpointState state)
+    {
+        // do nothing
+    }
+
+    public void onRemove(InetAddress endpoint)
+    {
+        // do nothing.
+    }
+
+    private void reConnect(InetAddress endpoint, VersionedValue versionedValue)
+    {
+        if (!getDatacenter(endpoint).equals(myDC))
+            return; // do nothing return back...
+
+        try
+        {
+            InetAddress remoteIP = InetAddress.getByName(versionedValue.value);
+            MessagingService.instance().getConnectionPool(endpoint).reset(remoteIP);
+            logger.debug(String.format("Intiated reconnect to an Internal IP %s for the endpoint %s", remoteIP, endpoint));
+        }
+        catch (UnknownHostException e)
+        {
+            logger.error("Error in getting the IP address resolved", e);
+        }
+    }
+
+    @Override
+    public void gossiperStarting()
+    {
+        super.gossiperStarting();
+        Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_IP,
+                                                   StorageService.instance.valueFactory.internalIP(FBUtilities.getLocalAddress().getHostAddress()));
+        Gossiper.instance.register(this);
+    }
 }