JAVA-1388: Add dynamic port discovery for system.peers_v2 (apache#1065)

Author: tolbertam

changelog/README.md

@@ -17,6 +17,7 @@
 - [bug] JAVA-1928: Fix GuavaCompatibility for Guava 26.
 - [bug] JAVA-1935: Add null check in QueryConsistencyException.getHost.
 - [improvement] JAVA-1771: Send driver name and version in STARTUP message.
+- [improvement] JAVA-1388: Add dynamic port discovery for system.peers\_v2.
 
 Merged from 3.5.x:
 

driver-core/src/main/java/com/datastax/driver/core/Cluster.java

@@ -1686,11 +1686,15 @@ ReconnectionPolicy reconnectionPolicy() {
       return configuration.getPolicies().getReconnectionPolicy();
     }
 
+    InetSocketAddress translateAddress(InetSocketAddress address) {
+      InetSocketAddress translated =
+          configuration.getPolicies().getAddressTranslator().translate(address);
+      return translated == null ? address : translated;
+    }
+
     InetSocketAddress translateAddress(InetAddress address) {
       InetSocketAddress sa = new InetSocketAddress(address, connectionFactory.getPort());
-      InetSocketAddress translated =
-          configuration.getPolicies().getAddressTranslator().translate(sa);
-      return translated == null ? sa : translated;
+      return translateAddress(sa);
     }
 
     private Session newSession() {

driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java

@@ -44,12 +44,12 @@ public class Host {
   // We use that internally because
   // that's the 'peer' in the 'System.peers' table and avoids querying the full peers table in
   // ControlConnection.refreshNodeInfo.
-  private volatile InetAddress broadcastAddress;
+  private volatile InetSocketAddress broadcastSocketAddress;
 
   // The listen_address as known by Cassandra.
   // This is usually the same as broadcast_address unless
   // specified otherwise in cassandra.yaml file.
-  private volatile InetAddress listenAddress;
+  private volatile InetSocketAddress listenSocketAddress;
 
   private volatile UUID hostId;
 
@@ -118,12 +118,12 @@ void setVersion(String cassandraVersion) {
     this.cassandraVersion = versionNumber;
   }
 
-  void setBroadcastAddress(InetAddress broadcastAddress) {
-    this.broadcastAddress = broadcastAddress;
+  void setBroadcastSocketAddress(InetSocketAddress broadcastAddress) {
+    this.broadcastSocketAddress = broadcastAddress;
   }
 
-  void setListenAddress(InetAddress listenAddress) {
-    this.listenAddress = listenAddress;
+  void setListenSocketAddress(InetSocketAddress listenAddress) {
+    this.listenSocketAddress = listenAddress;
   }
 
   void setDseVersion(String dseVersion) {
@@ -194,12 +194,30 @@ public InetSocketAddress getSocketAddress() {
   }
 
   /**
-   * Returns the node broadcast address (that is, the IP by which it should be contacted by other
-   * peers in the cluster), if known.
+   * Returns the node broadcast address, if known. Otherwise {@code null}.
+   *
+   * <p>This is a shortcut for {@code getBroadcastSocketAddress().getAddress()}.
+   *
+   * @return the node broadcast address, if known. Otherwise {@code null}.
+   * @see #getBroadcastSocketAddress()
+   * @see <a
+   *     href="https://docs.datastax.com/en/cassandra/2.1/cassandra/configuration/configCassandra_yaml_r.html">The
+   *     cassandra.yaml configuration file</a>
+   */
+  public InetAddress getBroadcastAddress() {
+    return broadcastSocketAddress != null ? broadcastSocketAddress.getAddress() : null;
+  }
+
+  /**
+   * Returns the node broadcast address (that is, the address by which it should be contacted by
+   * other peers in the cluster), if known. Otherwise {@code null}.
+   *
+   * <p>Note that the port of the returned address will be 0 for versions of Cassandra older than
+   * 4.0.
    *
    * <p>This corresponds to the {@code broadcast_address} cassandra.yaml file setting and is by
-   * default the same as {@link #getListenAddress()}, unless specified otherwise in cassandra.yaml.
-   * <em>This is NOT the address clients should use to contact this node</em>.
+   * default the same as {@link #getListenSocketAddress()}, unless specified otherwise in
+   * cassandra.yaml. <em>This is NOT the address clients should use to contact this node</em>.
    *
    * <p>This information is always available for peer hosts. For the control host, it's only
    * available if CASSANDRA-9436 is fixed on the server side (Cassandra versions >= 2.0.16, 2.1.6,
@@ -212,13 +230,31 @@ public InetSocketAddress getSocketAddress() {
    *     href="https://docs.datastax.com/en/cassandra/2.1/cassandra/configuration/configCassandra_yaml_r.html">The
    *     cassandra.yaml configuration file</a>
    */
-  public InetAddress getBroadcastAddress() {
-    return broadcastAddress;
+  public InetSocketAddress getBroadcastSocketAddress() {
+    return broadcastSocketAddress;
   }
 
   /**
-   * Returns the node listen address (that is, the IP the node uses to contact other peers in the
-   * cluster), if known.
+   * Returns the node listen address, if known. Otherwise {@code null}.
+   *
+   * <p>This is a shortcut for {@code getListenSocketAddress().getAddress()}.
+   *
+   * @return the node listen address, if known. Otherwise {@code null}.
+   * @see #getListenSocketAddress()
+   * @see <a
+   *     href="https://docs.datastax.com/en/cassandra/2.1/cassandra/configuration/configCassandra_yaml_r.html">The
+   *     cassandra.yaml configuration file</a>
+   */
+  public InetAddress getListenAddress() {
+    return listenSocketAddress != null ? listenSocketAddress.getAddress() : null;
+  }
+
+  /**
+   * Returns the node listen address (that is, the address the node uses to contact other peers in
+   * the cluster), if known. Otherwise {@code null}.
+   *
+   * <p>Note that the port of the returned address will be 0 for versions of Cassandra older than
+   * 4.0.
    *
    * <p>This corresponds to the {@code listen_address} cassandra.yaml file setting. <em>This is NOT
    * the address clients should use to contact this node</em>.
@@ -234,8 +270,8 @@ public InetAddress getBroadcastAddress() {
    *     href="https://docs.datastax.com/en/cassandra/2.1/cassandra/configuration/configCassandra_yaml_r.html">The
    *     cassandra.yaml configuration file</a>
    */
-  public InetAddress getListenAddress() {
-    return listenAddress;
+  public InetSocketAddress getListenSocketAddress() {
+    return listenSocketAddress;
   }
 
   /**

driver-core/src/main/java/com/datastax/driver/core/Host.java

@@ -15,9 +15,11 @@
  */
 package com.datastax.driver.core.utils;
 
+import com.datastax.driver.core.GuavaCompatibility;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 
 /** Helpers to work with Guava's {@link ListenableFuture}. */
 public class MoreFutures {
@@ -39,4 +41,28 @@ public void onSuccess(V result) {
       /* nothing */
     }
   }
+
+  /**
+   * Configures a {@link SettableFuture} to propagate the result of a future.
+   *
+   * @param settable future to be propagated to
+   * @param future future to propagate
+   * @param <T>
+   */
+  public static <T> void propagateFuture(
+      final SettableFuture<T> settable, ListenableFuture<T> future) {
+    GuavaCompatibility.INSTANCE.addCallback(
+        future,
+        new FutureCallback<T>() {
+          @Override
+          public void onSuccess(T result) {
+            settable.set(result);
+          }
+
+          @Override
+          public void onFailure(Throwable t) {
+            settable.setException(t);
+          }
+        });
+  }
 }

driver-core/src/main/java/com/datastax/driver/core/utils/MoreFutures.java

@@ -44,6 +44,7 @@
 import org.scassandra.Scassandra;
 import org.scassandra.http.client.PrimingClient;
 import org.scassandra.http.client.PrimingRequest;
+import org.scassandra.http.client.Result;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.Test;
@@ -310,5 +311,12 @@ private void primePeerRows(Scassandra scassandra, List<FakeHost> otherHosts)
             .withQuery("SELECT * FROM system.peers")
             .withThen(then().withRows(rows).withColumnTypes(ScassandraCluster.SELECT_PEERS))
             .build());
+
+    // prime invalid for peers_v2 so peers table is used.
+    primingClient.prime(
+        PrimingRequest.queryBuilder()
+            .withQuery("SELECT * FROM system.peers_v2")
+            .withThen(then().withResult(Result.invalid))
+            .build());
   }
 }