Review fixes

To be squashed into previous before merge.

Author: Bouncheck

core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java

@@ -278,7 +278,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) {
     map.put(TypedDriverOption.CONNECTION_WARN_INIT_ERROR, true);
     map.put(TypedDriverOption.CONNECTION_ADVANCED_SHARD_AWARENESS_ENABLED, true);
     map.put(TypedDriverOption.ADVANCED_SHARD_AWARENESS_PORT_LOW, 10000);
-    map.put(TypedDriverOption.ADVANCED_SHARD_AWARENESS_PORT_HIGH, 60000);
+    map.put(TypedDriverOption.ADVANCED_SHARD_AWARENESS_PORT_HIGH, 65535);
     map.put(TypedDriverOption.RECONNECT_ON_INIT, false);
     map.put(TypedDriverOption.RECONNECTION_POLICY_CLASS, "ExponentialReconnectionPolicy");
     map.put(TypedDriverOption.RECONNECTION_BASE_DELAY, Duration.ofSeconds(1));

core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelFactory.java

@@ -478,7 +478,7 @@ public static int getNextAvailablePort(int shardCount, int shardId, DriverContex
 
         // Scan from scanStart upwards to highPort.
         for (int port = scanStart; port <= highPort; port += shardCount) {
-          if (isTcpPortAvailable(port)) {
+          if (isTcpPortAvailable(port, context)) {
             foundPort = port;
             break;
           }
@@ -492,7 +492,7 @@ public static int getNextAvailablePort(int shardCount, int shardId, DriverContex
           scanStart = lowPort + (shardCount - lowPort % shardCount) + shardId;
 
           for (int port = scanStart; port <= highPort; port += shardCount) {
-            if (isTcpPortAvailable(port)) {
+            if (isTcpPortAvailable(port, context)) {
               foundPort = port;
               break;
             }
@@ -508,11 +508,15 @@ public static int getNextAvailablePort(int shardCount, int shardId, DriverContex
       return foundPort;
     }
 
-    public static boolean isTcpPortAvailable(int port) {
+    public static boolean isTcpPortAvailable(int port, DriverContext context) {
       try {
         ServerSocket serverSocket = new ServerSocket();
         try {
-          serverSocket.setReuseAddress(false);
+          serverSocket.setReuseAddress(
+              context
+                  .getConfig()
+                  .getDefaultProfile()
+                  .getBoolean(DefaultDriverOption.SOCKET_REUSE_ADDRESS, false));
           serverSocket.bind(new InetSocketAddress(port), 1);
           return true;
         } finally {

core/src/main/resources/reference.conf

@@ -540,15 +540,24 @@ datastax-java-driver {
     advanced-shard-awareness {
       # Whether to use advanced shard awareness when trying to open new connections.
       #
+
+      # Requires passing shard-aware port as contact point (usually 19042 or 19142(ssl)).
       # Having this enabled makes sense only for ScyllaDB clusters.
       # Results in smaller connection storms in multi-client settings.
-      # If set to false the driver will not attempt to use this feature.
+      # In short it's a feature that allows targeting particular shard when connecting to a node by using specific
+      # local port number.
+      # For context see https://www.scylladb.com/2021/04/27/connect-faster-to-scylla-with-a-shard-aware-port/
+      #
+      # If set to false the driver will not attempt to use this feature. This means connection's local port
+      # will be random according to system rules and driver will keep opening connections until it gets right shards.
+      # In such case non-shard aware port is recommended (by default 9042 or 9142).
       # If set to true the driver will attempt to use it and will log warnings each time something
       # makes it not possible.
+      #
       # If the node for some reason does not report it's sharding info the driver
       # will log a warning and create connection the same way as if this feature was disabled.
       # If the cluster ignores the request for specific shard warning will also be logged,
-      # although the local port will be chosen according to advanced shard awareness rules.
+      # although the local port will already be chosen according to advanced shard awareness rules.
       #
       # Required: yes
       # Modifiable at runtime: yes, the new value will be used for connections created after the
@@ -561,14 +570,16 @@ datastax-java-driver {
       # Required: yes
       # Modifiable at runtime: yes, the new value will be used for calls after the
       # change.
+      # Overridable in a profile: no
       port-low = 10000
 
       # Inclusive upper bound of port range to use in advanced shard awareness.
       # The driver will attempt to reserve ports for connection only within the range.
       # Required: yes
       # Modifiable at runtime: yes, the new value will be used for calls after the
       # change.
-      port-high = 60000
+      # Overridable in a profile: no
+      port-high = 65535
     }
   }
 

integration-tests/src/test/java/com/datastax/oss/driver/core/pool/AdvancedShardAwarenessIT.java

@@ -20,6 +20,8 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.Uninterruptibles;
+import com.tngtech.java.junit.dataprovider.DataProvider;
+import com.tngtech.java.junit.dataprovider.UseDataProvider;
 import java.net.InetSocketAddress;
 import java.time.Duration;
 import java.util.List;
@@ -54,6 +56,11 @@ public class AdvancedShardAwarenessIT {
       Pattern.compile(".*Scheduling next reconnection in.*");
   Set<Pattern> forbiddenOccurences = ImmutableSet.of(shardMismatchPattern, reconnectionPattern);
 
+  @DataProvider
+  public static Object[][] reuseAddressOption() {
+    return new Object[][] {{true}, {false}};
+  }
+
   @Before
   public void startCapturingLogs() {
     originalLevelChannelPool = channelPoolLogger.getLevel();
@@ -80,7 +87,8 @@ public void stopCapturingLogs() {
   }
 
   @Test
-  public void should_initialize_all_channels() {
+  @UseDataProvider("reuseAddressOption")
+  public void should_initialize_all_channels(boolean reuseAddress) {
     Map<Pattern, Integer> expectedOccurences =
         ImmutableMap.of(
             Pattern.compile(
@@ -96,6 +104,7 @@ public void should_initialize_all_channels() {
             Pattern.compile(".*127\\.0\\.0\\.2:19042\\] Trying to create 5 missing channels.*"), 1);
     DriverConfigLoader loader =
         SessionUtils.configLoaderBuilder()
+            .withBoolean(DefaultDriverOption.SOCKET_REUSE_ADDRESS, reuseAddress)
             .withBoolean(DefaultDriverOption.CONNECTION_ADVANCED_SHARD_AWARENESS_ENABLED, true)
             .withInt(DefaultDriverOption.ADVANCED_SHARD_AWARENESS_PORT_LOW, 10000)
             .withInt(DefaultDriverOption.ADVANCED_SHARD_AWARENESS_PORT_HIGH, 60000)