fix broadcastAddress and Ec2MultiRegionSnitch.reconnect

patch by Vijay; reviewed by Brandon Williams for CASSANDRA-3835

src/java/org/apache/cassandra/cli/CliClient.java

@@ -1155,15 +1155,7 @@ public class CliClient
             {
                 SimpleSnitch snitch = new SimpleSnitch();
                 Map<String, String> options = new HashMap<String, String>();
-
-                try
-                {
-                    options.put(snitch.getDatacenter(InetAddress.getLocalHost()), "1");
-                }
-                catch (UnknownHostException e)
-                {
-                    throw new RuntimeException(e);
-                }
+                options.put(snitch.getDatacenter(FBUtilities.getBroadcastAddress()), "1");
 
                 ksDef.setStrategy_options(options);
             }

src/java/org/apache/cassandra/db/SystemTable.java

@@ -138,7 +138,7 @@ public class SystemTable
      */
     public static synchronized void updateToken(InetAddress ep, Token token)
     {
-        if (ep == FBUtilities.getLocalAddress())
+        if (ep == FBUtilities.getBroadcastAddress())
         {
             removeToken(token);
             return;

src/java/org/apache/cassandra/gms/Gossiper.java

@@ -1095,7 +1095,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
      */
     public void addSavedEndpoint(InetAddress ep)
     {
-        if (ep == FBUtilities.getLocalAddress())
+        if (ep == FBUtilities.getBroadcastAddress())
         {
             logger.debug("Attempt to add self as saved endpoint");
             return;

src/java/org/apache/cassandra/net/OutboundTcpConnection.java

@@ -78,6 +78,11 @@ public class OutboundTcpConnection extends Thread
         enqueue(CLOSE_SENTINEL, null);
     }
 
+    void softCloseSocket()
+    {
+        enqueue(CLOSE_SENTINEL, null);
+    }
+
     public void run()
     {
         while (true)

src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java

@@ -68,7 +68,8 @@ public class OutboundTcpConnectionPool
     public void reset(InetAddress remoteEP)
     {
         resetedEndpoint = remoteEP;
-        reset();
+        for (OutboundTcpConnection con : new OutboundTcpConnection[] { cmdCon, ackCon })
+            con.softCloseSocket();
     }
     
     public Socket newSocket() throws IOException
@@ -78,7 +79,8 @@ public class OutboundTcpConnectionPool
         {
             return SSLFactory.getSocket(DatabaseDescriptor.getEncryptionOptions(), endPoint(), DatabaseDescriptor.getSSLStoragePort(), FBUtilities.getLocalAddress(), 0);
         }
-        else {
+        else 
+        {
             return new Socket(endPoint(), DatabaseDescriptor.getStoragePort(), FBUtilities.getLocalAddress(), 0);
         }
     }

src/java/org/apache/cassandra/service/MigrationManager.java

@@ -54,7 +54,7 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
     // avoids re-pushing migrations that we're waiting on target to apply already
     private static Map<InetAddress,UUID> lastPushed = new MapMaker().expiration(1, TimeUnit.MINUTES).makeMap();
     
-    private static UUID highestKnown;
+    private static volatile UUID highestKnown;
 
     public void onJoin(InetAddress endpoint, EndpointState epState) { 
         VersionedValue value = epState.getApplicationState(ApplicationState.SCHEMA);