JAVA-2967: Support native_transport_(address|port) + native_transport_port_ssl for DSE 6.8 (#1573)

Author: absurdfarce

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

@@ -611,6 +611,17 @@ private static void updateInfo(
       InetAddress nativeAddress = row.getInet("native_address");
       int nativePort = row.getInt("native_port");
       broadcastRpcAddress = new InetSocketAddress(nativeAddress, nativePort);
+    } else if (row.getColumnDefinitions().contains("native_transport_address")) {
+      // DSE 6.8 introduced native_transport_address and native_transport_port for the
+      // listen address.  Also included is native_transport_port_ssl (in case users
+      // want to setup a different port for SSL and non-SSL conns).
+      InetAddress nativeAddress = row.getInet("native_transport_address");
+      int nativePort = row.getInt("native_transport_port");
+      if (cluster.getCluster().getConfiguration().getProtocolOptions().getSSLOptions() != null
+          && !row.isNull("native_transport_port_ssl")) {
+        nativePort = row.getInt("native_transport_port_ssl");
+      }
+      broadcastRpcAddress = new InetSocketAddress(nativeAddress, nativePort);
     } else if (row.getColumnDefinitions().contains("rpc_address")) {
       InetAddress rpcAddress = row.getInet("rpc_address");
       broadcastRpcAddress = new InetSocketAddress(rpcAddress, cluster.connectionFactory.getPort());
@@ -831,6 +842,14 @@ private void refreshNodeListAndTokenMap(
         InetAddress nativeAddress = row.getInet("native_address");
         int nativePort = row.getInt("native_port");
         broadcastRpcAddress = new InetSocketAddress(nativeAddress, nativePort);
+      } else if (row.getColumnDefinitions().contains("native_transport_address")) {
+        InetAddress nativeAddress = row.getInet("native_transport_address");
+        int nativePort = row.getInt("native_transport_port");
+        if (cluster.getCluster().getConfiguration().getProtocolOptions().getSSLOptions() != null
+            && !row.isNull("native_transport_port_ssl")) {
+          nativePort = row.getInt("native_transport_port_ssl");
+        }
+        broadcastRpcAddress = new InetSocketAddress(nativeAddress, nativePort);
       } else {
         InetAddress rpcAddress = row.getInet("rpc_address");
         broadcastRpcAddress =
@@ -956,7 +975,11 @@ private boolean isValidPeer(Row peerRow, boolean logIfInvalid) {
               && !peerRow.isNull("native_port");
     } else {
       isValid &=
-          peerRow.getColumnDefinitions().contains("rpc_address") && !peerRow.isNull("rpc_address");
+          (peerRow.getColumnDefinitions().contains("rpc_address") && !peerRow.isNull("rpc_address"))
+              || (peerRow.getColumnDefinitions().contains("native_transport_address")
+                  && peerRow.getColumnDefinitions().contains("native_transport_port")
+                  && !peerRow.isNull("native_transport_address")
+                  && !peerRow.isNull("native_transport_port"));
     }
 
     if (EXTENDED_PEER_CHECK) {
@@ -984,6 +1007,9 @@ private String formatInvalidPeer(Row peerRow) {
       formatMissingOrNullColumn(peerRow, "native_address", sb);
       formatMissingOrNullColumn(peerRow, "native_port", sb);
     } else {
+      formatMissingOrNullColumn(peerRow, "native_transport_address", sb);
+      formatMissingOrNullColumn(peerRow, "native_transport_port", sb);
+      formatMissingOrNullColumn(peerRow, "native_transport_port_ssl", sb);
       formatMissingOrNullColumn(peerRow, "rpc_address", sb);
     }
     if (EXTENDED_PEER_CHECK) {

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

@@ -49,6 +49,16 @@ public EndPoint create(Row peersRow) {
       InetSocketAddress translateAddress =
           cluster.manager.translateAddress(new InetSocketAddress(nativeAddress, nativePort));
       return new TranslatedAddressEndPoint(translateAddress);
+    } else if (peersRow.getColumnDefinitions().contains("native_transport_address")) {
+      InetAddress nativeAddress = peersRow.getInet("native_transport_address");
+      int nativePort = peersRow.getInt("native_transport_port");
+      if (cluster.getConfiguration().getProtocolOptions().getSSLOptions() != null
+          && !peersRow.isNull("native_transport_port_ssl")) {
+        nativePort = peersRow.getInt("native_transport_port_ssl");
+      }
+      InetSocketAddress translateAddress =
+          cluster.manager.translateAddress(new InetSocketAddress(nativeAddress, nativePort));
+      return new TranslatedAddressEndPoint(translateAddress);
     } else {
       InetAddress broadcastAddress = peersRow.getInet("peer");
       InetAddress rpcAddress = peersRow.getInet("rpc_address");