move SeedProvider and AbstractNetworkTopologySnitch to InetAddressAndPort for forward compatibility

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -30,8 +30,8 @@
 import java.util.*;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
 import com.google.common.primitives.Ints;
 import com.google.common.primitives.Longs;
 
@@ -295,6 +295,8 @@ else if (config.rpc_interface != null)
     public static void applyConfig(Config config) throws ConfigurationException
     {
         conf = config;
+
+        InetAddressAndPort.initializeDefaultPort(getSSLStoragePort());
 
         if (conf.commitlog_sync == null)
         {
@@ -731,7 +733,8 @@ else if (conf.num_tokens > MAX_NUM_TOKENS)
         if (Boolean.getBoolean("palantir_cassandra.single_node_cluster"))
         {
             try {
-                if (!seedProvider.getSeeds().equals(ImmutableList.of(InetAddress.getLocalHost())))
+                List<InetAddressAndPort> seeds = seedProvider.getSeeds();
+                if (seeds.size() != 1 || !Iterables.getOnlyElement(seeds).address.equals(InetAddress.getLocalHost()))
                     throw new ConfigurationException("Unexpected seed list when single_node_cluster flag is set to true."
                         + " For a single node Cassandra cluster, the only seed should be localhost", false);
             } catch (UnknownHostException e) {
@@ -1355,7 +1358,12 @@ public static String getPersistentSettingsLocation() {
 
     public static Set<InetAddress> getSeeds()
     {
-        return ImmutableSet.<InetAddress>builder().addAll(seedProvider.getSeeds()).build();
+        ImmutableSet.Builder<InetAddress> builder = ImmutableSet.builder();
+        for (InetAddressAndPort seed : seedProvider.getSeeds())
+        {
+            builder.add(seed.address);
+        }
+        return builder.build();
     }
 
     public static Set<InetAddress> getAllHosts()

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

@@ -30,6 +30,7 @@
 import com.google.common.io.ByteStreams;
 import com.palantir.cassandra.db.CompactionsInProgressFlusher;
 
+import org.apache.cassandra.locator.InetAddressAndPort;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -666,12 +667,12 @@ public static InetAddress getPreferredIP(InetAddress ep)
     /**
      * Return a map of IP addresses containing a map of dc and rack info
      */
-    public static Map<InetAddress, Map<String,String>> loadDcRackInfo()
+    public static Map<InetAddressAndPort, Map<String, String>> loadDcRackInfo()
     {
-        Map<InetAddress, Map<String, String>> result = new HashMap<>();
+        Map<InetAddressAndPort, Map<String, String>> result = new HashMap<>();
         for (UntypedResultSet.Row row : executeInternal("SELECT peer, data_center, rack from system." + PEERS))
         {
-            InetAddress peer = row.getInetAddress("peer");
+            InetAddressAndPort peer = InetAddressAndPort.getByAddress(row.getInetAddress("peer"));
             if (row.has("data_center") && row.has("rack"))
             {
                 Map<String, String> dcRack = new HashMap<>();

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

@@ -31,6 +31,7 @@
 import com.google.common.util.concurrent.Uninterruptibles;
 
 import com.palantir.cassandra.cvim.CrossVpcIpMappingHandshaker;
+import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.utils.ExecutorUtils;
 import org.apache.cassandra.utils.MBeanWrapper;
 import org.apache.cassandra.utils.Pair;
@@ -858,6 +859,12 @@ protected long getExpireTimeForEndpoint(InetAddress endpoint)
         return storedTime == null ? computeExpireTime() : storedTime;
     }
 
+    public EndpointState getEndpointStateForEndpoint(InetAddressAndPort ep)
+    {
+        return getEndpointStateForEndpoint(ep.address);
+    }
+
+    @Deprecated
     public EndpointState getEndpointStateForEndpoint(InetAddress ep)
     {
         return endpointStateMap.get(ep);

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

@@ -30,14 +30,24 @@ public abstract class AbstractNetworkTopologySnitch extends AbstractEndpointSnit
      * @param endpoint a specified endpoint
      * @return string of rack
      */
-    abstract public String getRack(InetAddress endpoint);
+    abstract public String getRack(InetAddressAndPort endpoint);
+
+    public String getRack(InetAddress endpoint)
+    {
+        return getRack(InetAddressAndPort.getByAddress(endpoint));
+    }
 
     /**
      * Return the data center for which an endpoint resides in
      * @param endpoint a specified endpoint
      * @return string of data center
      */
-    abstract public String getDatacenter(InetAddress endpoint);
+    abstract public String getDatacenter(InetAddressAndPort endpoint);
+
+    public String getDatacenter(InetAddress endpoint)
+    {
+        return getDatacenter(InetAddressAndPort.getByAddress(endpoint));
+    }
 
     public int compareEndpoints(InetAddress address, InetAddress a1, InetAddress a2)
     {

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

@@ -56,7 +56,7 @@ public class CloudstackSnitch extends AbstractNetworkTopologySnitch
     protected static final Logger logger = LoggerFactory.getLogger(CloudstackSnitch.class);
     protected static final String ZONE_NAME_QUERY_URI = "/latest/meta-data/availability-zone";
 
-    private Map<InetAddress, Map<String, String>> savedEndpoints;
+    private Map<InetAddressAndPort, Map<String, String>> savedEndpoints;
 
     private static final String DEFAULT_DC = "UNKNOWN-DC";
     private static final String DEFAULT_RACK = "UNKNOWN-RACK";
@@ -83,9 +83,9 @@ public CloudstackSnitch() throws IOException, ConfigurationException
         csZoneRack = zone_parts[2];
     }
 
-    public String getRack(InetAddress endpoint)
+    public String getRack(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return csZoneRack;
         EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
         if (state == null || state.getApplicationState(ApplicationState.RACK) == null) 
@@ -99,9 +99,9 @@ public String getRack(InetAddress endpoint)
         return state.getApplicationState(ApplicationState.RACK).value;
     }
 
-    public String getDatacenter(InetAddress endpoint)
+    public String getDatacenter(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return csZoneDc;
         EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
         if (state == null || state.getApplicationState(ApplicationState.DC) == null) 

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

@@ -46,7 +46,7 @@ public class Ec2Snitch extends AbstractNetworkTopologySnitch
     protected static final String ZONE_NAME_QUERY_URL = "http://169.254.169.254/latest/meta-data/placement/availability-zone";
     private static final String DEFAULT_DC = "UNKNOWN-DC";
     private static final String DEFAULT_RACK = "UNKNOWN-RACK";
-    private Map<InetAddress, Map<String, String>> savedEndpoints;
+    private Map<InetAddressAndPort, Map<String, String>> savedEndpoints;
     protected String ec2zone;
     protected String ec2region;
 
@@ -92,9 +92,9 @@ String awsApiCall(String url) throws IOException, ConfigurationException
         }
     }
 
-    public String getRack(InetAddress endpoint)
+    public String getRack(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return ec2zone;
         EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
         if (state == null || state.getApplicationState(ApplicationState.RACK) == null)
@@ -108,9 +108,9 @@ public String getRack(InetAddress endpoint)
         return state.getApplicationState(ApplicationState.RACK).value;
     }
 
-    public String getDatacenter(InetAddress endpoint)
+    public String getDatacenter(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return ec2region;
         EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
         if (state == null || state.getApplicationState(ApplicationState.DC) == null)

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

@@ -46,7 +46,7 @@ public class GoogleCloudSnitch extends AbstractNetworkTopologySnitch
     protected static final String ZONE_NAME_QUERY_URL = "http://metadata.google.internal/computeMetadata/v1/instance/zone";
     private static final String DEFAULT_DC = "UNKNOWN-DC";
     private static final String DEFAULT_RACK = "UNKNOWN-RACK";
-    private Map<InetAddress, Map<String, String>> savedEndpoints;
+    private Map<InetAddressAndPort, Map<String, String>> savedEndpoints;
     protected String gceZone;
     protected String gceRegion;
 
@@ -94,9 +94,9 @@ String gceApiCall(String url) throws IOException, ConfigurationException
         }
     }
 
-    public String getRack(InetAddress endpoint)
+    public String getRack(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return gceZone;
         EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
         if (state == null || state.getApplicationState(ApplicationState.RACK) == null)
@@ -110,9 +110,9 @@ public String getRack(InetAddress endpoint)
         return state.getApplicationState(ApplicationState.RACK).value;
     }
 
-    public String getDatacenter(InetAddress endpoint)
+    public String getDatacenter(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return gceRegion;
         EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
         if (state == null || state.getApplicationState(ApplicationState.DC) == null)

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

@@ -45,7 +45,7 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch//
     private final boolean preferLocal;
     private final AtomicReference<ReconnectableSnitchHelper> snitchHelperReference;
 
-    private Map<InetAddress, Map<String, String>> savedEndpoints;
+    private Map<InetAddressAndPort, Map<String, String>> savedEndpoints;
     private static final String DEFAULT_DC = "UNKNOWN_DC";
     private static final String DEFAULT_RACK = "UNKNOWN_RACK";
 
@@ -84,9 +84,9 @@ private static SnitchProperties loadConfiguration() throws ConfigurationExceptio
      * @param endpoint the endpoint to process
      * @return string of data center
      */
-    public String getDatacenter(InetAddress endpoint)
+    public String getDatacenter(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return myDC;
 
         EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
@@ -112,9 +112,9 @@ public String getDatacenter(InetAddress endpoint)
      * @param endpoint the endpoint to process
      * @return string of rack
      */
-    public String getRack(InetAddress endpoint)
+    public String getRack(InetAddressAndPort endpoint)
     {
-        if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+        if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
             return myRack;
 
         EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);

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

@@ -49,7 +49,7 @@ public final class InetAddressAndPort implements Comparable<InetAddressAndPort>,
     //these when it loads the config. A lot of unit tests won't end up loading DatabaseDescriptor.
     //Tools that might use this class also might not load database descriptor. Those tools are expected
     //to always override the defaults.
-    static volatile int defaultPort = 7000;
+    static volatile int defaultPort = 7001;
 
     public final InetAddress address;
     public final byte[] addressBytes;
@@ -191,4 +191,9 @@ public static void initializeDefaultPort(int port)
     {
         defaultPort = port;
     }
+
+    public static InetAddressAndPort getLocalHost()
+    {
+        return FBUtilities.getLocalAddressAndPort();
+    }
 }

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

@@ -118,10 +118,10 @@ private static String[] getRawEndpointInfo(InetAddress endpoint)
      * @param endpoint the endpoint to process
      * @return string of data center
      */
-    public String getDatacenter(InetAddress endpoint)
+    public String getDatacenter(InetAddressAndPort endpoint)
     {
-        String[] info = getEndpointInfo(endpoint);
-        assert info != null : "No location defined for endpoint " + endpoint;
+        String[] info = getEndpointInfo(endpoint.address);
+        assert info != null : "No location defined for endpoint " + endpoint.address;
         return info[0];
     }
 
@@ -131,10 +131,10 @@ public String getDatacenter(InetAddress endpoint)
      * @param endpoint the endpoint to process
      * @return string of rack
      */
-    public String getRack(InetAddress endpoint)
+    public String getRack(InetAddressAndPort endpoint)
     {
-        String[] info = getEndpointInfo(endpoint);
-        assert info != null : "No location defined for endpoint " + endpoint;
+        String[] info = getEndpointInfo(endpoint.address);
+        assert info != null : "No location defined for endpoint " + endpoint.address;
         return info[1];
     }
 

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

@@ -25,13 +25,13 @@
  */
 public class RackInferringSnitch extends AbstractNetworkTopologySnitch
 {
-    public String getRack(InetAddress endpoint)
+    public String getRack(InetAddressAndPort endpoint)
     {
-        return Integer.toString(endpoint.getAddress()[2] & 0xFF, 10);
+        return Integer.toString(endpoint.address.getAddress()[2] & 0xFF, 10);
     }
 
-    public String getDatacenter(InetAddress endpoint)
+    public String getDatacenter(InetAddressAndPort endpoint)
     {
-        return Integer.toString(endpoint.getAddress()[1] & 0xFF, 10);
+        return Integer.toString(endpoint.address.getAddress()[1] & 0xFF, 10);
     }
 }

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

@@ -22,5 +22,5 @@
 
 public interface SeedProvider
 {
-    List<InetAddress> getSeeds();
+    List<InetAddressAndPort> getSeeds();
 }

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

@@ -35,7 +35,7 @@ public class SimpleSeedProvider implements SeedProvider
 
     public SimpleSeedProvider(Map<String, String> args) {}
 
-    public List<InetAddress> getSeeds()
+    public List<InetAddressAndPort> getSeeds()
     {
         Config conf;
         try
@@ -47,12 +47,12 @@ public List<InetAddress> getSeeds()
             throw new AssertionError(e);
         }
         String[] hosts = conf.seed_provider.parameters.get("seeds").split(",", -1);
-        List<InetAddress> seeds = new ArrayList<InetAddress>(hosts.length);
+        List<InetAddressAndPort> seeds = new ArrayList<>(hosts.length);
         for (String host : hosts)
         {
             try
             {
-                seeds.add(InetAddress.getByName(host.trim()));
+                seeds.add(InetAddressAndPort.getByName(host.trim()));
             }
             catch (UnknownHostException ex)
             {

src/java/org/apache/cassandra/utils/FBUtilities.java

@@ -143,6 +143,25 @@ public static InetAddress getLocalAddress()
         return localInetAddress;
     }
 
+    /**
+     * Get the broadcast address and port for intra-cluster storage traffic. This the address to advertise that uniquely
+     * identifies the node and is reachable from everywhere. This is the one you want unless you are trying to connect
+     * to the local address specifically.
+     */
+    public static InetAddressAndPort getBroadcastAddressAndPort()
+    {
+        return InetAddressAndPort.getByAddress(getBroadcastAddress());
+    }
+
+    /**
+     * The address and port to listen on for intra-cluster storage traffic (not client). Use this to get the correct
+     * stuff to listen on for intra-cluster communication.
+     */
+    public static InetAddressAndPort getLocalAddressAndPort()
+    {
+        return InetAddressAndPort.getByAddress(getLocalAddress());
+    }
+
     public static InetAddress getBroadcastAddress()
     {
         if (broadcastInetAddress == null)

test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java

@@ -105,7 +105,7 @@ private String maybeGetFromEndpointState(String current, InetAddressAndPort endp
             {
                 savedEndpoints = new HashMap<>();
                 int storage_port = Config.getOverrideLoadConfig().get().storage_port;
-                for (Map.Entry<InetAddress, Map<String, String>> entry : SystemKeyspace.loadDcRackInfo().entrySet())
+                for (Map.Entry<InetAddressAndPort, Map<String, String>> entry : SystemKeyspace.loadDcRackInfo().entrySet())
                 {
                     savedEndpoints.put(InetAddressAndPort.getByAddressOverrideDefaults(endpoint.address, storage_port),
                                        entry.getValue());