JAVA-3131: Add #retrieve method to EndPoint for when caller does not need the endpoint to be proactively resolved

Refactor existing usages of EndPoint#resolve to use retrieve when resolved ip addresses are not needed.

Author: hhughes

core/src/main/java/com/datastax/dse/driver/api/core/auth/DseGssApiAuthProviderBase.java

@@ -319,7 +319,7 @@ protected GssApiAuthenticator(
                 SUPPORTED_MECHANISMS,
                 options.getAuthorizationId(),
                 protocol,
-                ((InetSocketAddress) endPoint.resolve()).getAddress().getCanonicalHostName(),
+                ((InetSocketAddress) endPoint.retrieve()).getAddress().getCanonicalHostName(),
                 options.getSaslProperties(),
                 null);
       } catch (LoginException | SaslException e) {

core/src/main/java/com/datastax/dse/driver/internal/core/insights/InsightsClient.java

@@ -294,7 +294,7 @@ private Map<String, SessionStateForNode> getConnectedNodes() {
     return pools.entrySet().stream()
         .collect(
             Collectors.toMap(
-                entry -> AddressFormatter.nullSafeToString(entry.getKey().getEndPoint().resolve()),
+                entry -> AddressFormatter.nullSafeToString(entry.getKey().getEndPoint().retrieve()),
                 this::constructSessionStateForNode));
   }
 
@@ -315,7 +315,7 @@ private InsightsStartupData createStartupData() {
         .withContactPoints(
             getResolvedContactPoints(
                 driverContext.getMetadataManager().getContactPoints().stream()
-                    .map(n -> n.getEndPoint().resolve())
+                    .map(n -> n.getEndPoint().retrieve())
                     .filter(InetSocketAddress.class::isInstance)
                     .map(InetSocketAddress.class::cast)
                     .collect(Collectors.toSet())))
@@ -456,7 +456,7 @@ private PoolSizeByHostDistance getPoolSizeByHostDistance() {
   }
 
   private String getControlConnectionSocketAddress() {
-    SocketAddress controlConnectionAddress = controlConnection.channel().getEndPoint().resolve();
+    SocketAddress controlConnectionAddress = controlConnection.channel().getEndPoint().retrieve();
     return AddressFormatter.nullSafeToString(controlConnectionAddress);
   }
 

core/src/main/java/com/datastax/oss/driver/api/core/metadata/EndPoint.java

@@ -40,6 +40,17 @@ public interface EndPoint {
   @NonNull
   SocketAddress resolve();
 
+  /**
+   * Returns a possibly unresolved instance to a socket address.
+   *
+   * <p>This should be called when the address does not need to be proactively resolved. For example
+   * if the node hostname or port number is needed.
+   */
+  @NonNull
+  default SocketAddress retrieve() {
+    return resolve();
+  }
+
   /**
    * Returns an alternate string representation for use in node-level metric names.
    *

core/src/main/java/com/datastax/oss/driver/api/core/ssl/ProgrammaticSslEngineFactory.java

@@ -89,7 +89,7 @@ public ProgrammaticSslEngineFactory(
   @Override
   public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) {
     SSLEngine engine;
-    SocketAddress remoteAddress = remoteEndpoint.resolve();
+    SocketAddress remoteAddress = remoteEndpoint.retrieve();
     if (remoteAddress instanceof InetSocketAddress) {
       InetSocketAddress socketAddress = (InetSocketAddress) remoteAddress;
       engine = sslContext.createSSLEngine(socketAddress.getHostName(), socketAddress.getPort());

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPoint.java

@@ -38,6 +38,11 @@ public DefaultEndPoint(InetSocketAddress address) {
   @NonNull
   @Override
   public InetSocketAddress resolve() {
+    return retrieve();
+  }
+
+  @Override
+  public InetSocketAddress retrieve() {
     return address;
   }
 

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultTopologyMonitor.java

@@ -472,7 +472,7 @@ private Optional<NodeInfo> findInPeers(
   // We save it the first time we get a control connection channel.
   private void savePort(DriverChannel channel) {
     if (port < 0) {
-      SocketAddress address = channel.getEndPoint().resolve();
+      SocketAddress address = channel.getEndPoint().retrieve();
       if (address instanceof InetSocketAddress) {
         port = ((InetSocketAddress) address).getPort();
       }
@@ -518,7 +518,7 @@ protected InetSocketAddress getBroadcastRpcAddress(
     }
     InetSocketAddress broadcastRpcAddress =
         new InetSocketAddress(broadcastRpcInetAddress, broadcastRpcPort);
-    if (row.contains("peer") && broadcastRpcAddress.equals(localEndPoint.resolve())) {
+    if (row.contains("peer") && broadcastRpcAddress.equals(localEndPoint.retrieve())) {
       // JAVA-2303: if the peer is actually the control node, ignore that peer as it is likely
       // a misconfiguration problem.
       LOG.warn(

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java

@@ -18,6 +18,7 @@
 package com.datastax.oss.driver.internal.core.metadata;
 
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
 import com.datastax.oss.driver.shaded.guava.common.primitives.UnsignedBytes;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.net.InetAddress;
@@ -26,10 +27,14 @@
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class SniEndPoint implements EndPoint {
-  private static final AtomicLong OFFSET = new AtomicLong();
+  // initialize offset to random position to avoid all clients starting at the same index
+  @VisibleForTesting
+  static final AtomicInteger OFFSET =
+      new AtomicInteger(ThreadLocalRandom.current().nextInt(0, 1024));
 
   private final InetSocketAddress proxyAddress;
   private final String serverName;
@@ -55,7 +60,7 @@ public String getServerName() {
   @Override
   public InetSocketAddress resolve() {
     try {
-      InetAddress[] aRecords = InetAddress.getAllByName(proxyAddress.getHostName());
+      InetAddress[] aRecords = resolveARecords();
       if (aRecords.length == 0) {
         // Probably never happens, but the JDK docs don't explicitly say so
         throw new IllegalArgumentException(
@@ -64,14 +69,32 @@ public InetSocketAddress resolve() {
       // The order of the returned address is unspecified. Sort by IP to make sure we get a true
       // round-robin
       Arrays.sort(aRecords, IP_COMPARATOR);
-      int index = (aRecords.length == 1) ? 0 : (int) OFFSET.getAndIncrement() % aRecords.length;
-      return new InetSocketAddress(aRecords[index], proxyAddress.getPort());
+
+      // get next offset value, reset OFFSET if wrapped around to negative
+      int nextOffset = OFFSET.getAndIncrement();
+      if (nextOffset < 0) {
+        // if negative set OFFSET to 1 and nextOffset to 0, else simulate getAndIncrement()
+        nextOffset = OFFSET.updateAndGet(v -> v < 0 ? 1 : v + 1) - 1;
+      }
+
+      return new InetSocketAddress(aRecords[nextOffset % aRecords.length], proxyAddress.getPort());
     } catch (UnknownHostException e) {
       throw new IllegalArgumentException(
           "Could not resolve proxy address " + proxyAddress.getHostName(), e);
     }
   }
 
+  @VisibleForTesting
+  InetAddress[] resolveARecords() throws UnknownHostException {
+    // moving static call to method to allow mocking in tests
+    return InetAddress.getAllByName(proxyAddress.getHostName());
+  }
+
+  @Override
+  public InetSocketAddress retrieve() {
+    return proxyAddress;
+  }
+
   @Override
   public boolean equals(Object other) {
     if (other == this) {

core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java

@@ -90,7 +90,7 @@ public DefaultSslEngineFactory(DriverContext driverContext) {
   @Override
   public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) {
     SSLEngine engine;
-    SocketAddress remoteAddress = remoteEndpoint.resolve();
+    SocketAddress remoteAddress = remoteEndpoint.retrieve();
     if (remoteAddress instanceof InetSocketAddress) {
       InetSocketAddress socketAddress = (InetSocketAddress) remoteAddress;
       engine = sslContext.createSSLEngine(socketAddress.getHostName(), socketAddress.getPort());

core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java

@@ -53,7 +53,7 @@ public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) {
               this.getClass().getSimpleName()));
     }
     SniEndPoint sniEndPoint = (SniEndPoint) remoteEndpoint;
-    InetSocketAddress address = sniEndPoint.resolve();
+    InetSocketAddress address = sniEndPoint.retrieve();
     String sniServerName = sniEndPoint.getServerName();
 
     // When hostname verification is enabled (with setEndpointIdentificationAlgorithm), the SSL

core/src/test/java/com/datastax/dse/driver/internal/core/insights/InsightsClientTest.java

@@ -485,7 +485,7 @@ private DefaultDriverContext mockDefaultDriverContext() throws UnknownHostExcept
     when(context.getProtocolVersion()).thenReturn(DSE_V2);
     DefaultNode contactPoint = mock(DefaultNode.class);
     EndPoint contactEndPoint = mock(EndPoint.class);
-    when(contactEndPoint.resolve()).thenReturn(new InetSocketAddress("127.0.0.1", 9999));
+    when(contactEndPoint.retrieve()).thenReturn(new InetSocketAddress("127.0.0.1", 9999));
     when(contactPoint.getEndPoint()).thenReturn(contactEndPoint);
     when(manager.getContactPoints()).thenReturn(ImmutableSet.of(contactPoint));
 
@@ -501,7 +501,7 @@ private DefaultDriverContext mockDefaultDriverContext() throws UnknownHostExcept
     ControlConnection controlConnection = mock(ControlConnection.class);
     DriverChannel channel = mock(DriverChannel.class);
     EndPoint controlConnectionEndpoint = mock(EndPoint.class);
-    when(controlConnectionEndpoint.resolve()).thenReturn(new InetSocketAddress("127.0.0.1", 10));
+    when(controlConnectionEndpoint.retrieve()).thenReturn(new InetSocketAddress("127.0.0.1", 10));
 
     when(channel.getEndPoint()).thenReturn(controlConnectionEndpoint);
     when(channel.localAddress()).thenReturn(new InetSocketAddress("127.0.0.1", 10));
@@ -513,15 +513,15 @@ private DefaultDriverContext mockDefaultDriverContext() throws UnknownHostExcept
   private void mockConnectionPools(DefaultDriverContext driverContext) {
     Node node1 = mock(Node.class);
     EndPoint endPoint1 = mock(EndPoint.class);
-    when(endPoint1.resolve()).thenReturn(new InetSocketAddress("127.0.0.1", 10));
+    when(endPoint1.retrieve()).thenReturn(new InetSocketAddress("127.0.0.1", 10));
     when(node1.getEndPoint()).thenReturn(endPoint1);
     when(node1.getOpenConnections()).thenReturn(1);
     ChannelPool channelPool1 = mock(ChannelPool.class);
     when(channelPool1.getInFlight()).thenReturn(10);
 
     Node node2 = mock(Node.class);
     EndPoint endPoint2 = mock(EndPoint.class);
-    when(endPoint2.resolve()).thenReturn(new InetSocketAddress("127.0.0.1", 20));
+    when(endPoint2.retrieve()).thenReturn(new InetSocketAddress("127.0.0.1", 20));
     when(node2.getEndPoint()).thenReturn(endPoint2);
     when(node2.getOpenConnections()).thenReturn(2);
     ChannelPool channelPool2 = mock(ChannelPool.class);