[FEATURE] Cosmos | Wire up ConnectionPolicy, DirectConnectionConfig, and RntbdTransportClient.Options

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/AsyncBenchmark.java

@@ -131,7 +131,7 @@ abstract class AsyncBenchmark<T> {
         if (configuration.getOperationType() != Configuration.Operation.WriteLatency
                 && configuration.getOperationType() != Configuration.Operation.WriteThroughput
                 && configuration.getOperationType() != Configuration.Operation.ReadMyWrites) {
-            logger.info("PRE-populating {} documents ....", cfg.getNumberOfOperations());
+            logger.info("PRE-populating {} documents ....", cfg.getNumberOfPreCreatedDocuments());
             String dataFieldValue = RandomStringUtils.randomAlphabetic(cfg.getDocumentDataFieldSize());
             for (int i = 0; i < cfg.getNumberOfPreCreatedDocuments(); i++) {
                 String uuid = UUID.randomUUID().toString();
@@ -144,7 +144,7 @@ abstract class AsyncBenchmark<T> {
                 }).flux();
                 createDocumentObservables.add(obs);
             }
-            logger.info("Finished pre-populating {} documents", cfg.getNumberOfOperations());
+            logger.info("Finished pre-populating {} documents", cfg.getNumberOfPreCreatedDocuments());
         }
 
         docsToRead = Flux.merge(Flux.fromIterable(createDocumentObservables), 100).collectList().block();

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/DirectConnectionConfig.java

@@ -17,24 +17,27 @@ public final class DirectConnectionConfig {
     //  Constants
     private static final Duration DEFAULT_IDLE_ENDPOINT_TIMEOUT = Duration.ofSeconds(70L);
     private static final Duration DEFAULT_CONNECTION_TIMEOUT = Duration.ofSeconds(60L);
+    private static final Duration DEFAULT_REQUEST_TIMEOUT = Duration.ofSeconds(60L);
     private static final int DEFAULT_MAX_CONNECTIONS_PER_ENDPOINT = 30;
     private static final int DEFAULT_MAX_REQUESTS_PER_CONNECTION = 10;
 
     private Duration connectionTimeout;
     private Duration idleConnectionTimeout;
     private Duration idleEndpointTimeout;
+    private Duration requestTimeout;
     private int maxConnectionsPerEndpoint;
     private int maxRequestsPerConnection;
 
     /**
      * Constructor.
      */
     public DirectConnectionConfig() {
-        this.idleConnectionTimeout = Duration.ZERO;
         this.connectionTimeout = DEFAULT_CONNECTION_TIMEOUT;
+        this.idleConnectionTimeout = Duration.ZERO;
         this.idleEndpointTimeout = DEFAULT_IDLE_ENDPOINT_TIMEOUT;
         this.maxConnectionsPerEndpoint = DEFAULT_MAX_CONNECTIONS_PER_ENDPOINT;
         this.maxRequestsPerConnection = DEFAULT_MAX_REQUESTS_PER_CONNECTION;
+        this.requestTimeout = DEFAULT_REQUEST_TIMEOUT;
     }
 
     /**
@@ -190,6 +193,32 @@ public DirectConnectionConfig setMaxRequestsPerConnection(int maxRequestsPerConn
         return this;
     }
 
+    /**
+     * Gets the request timeout interval
+     * This represents the timeout interval for requests
+     *
+     * Default value is 60 seconds
+     *
+     * @return the request timeout interval
+     */
+    public Duration getRequestTimeout() {
+        return requestTimeout;
+    }
+
+    /**
+     * Sets the request timeout interval
+     * This represents the timeout interval for requests
+     *
+     * Default value is 60 seconds
+     *
+     * @param requestTimeout the request timeout interval
+     * @return the {@link DirectConnectionConfig}
+     */
+    public DirectConnectionConfig setRequestTimeout(Duration requestTimeout) {
+        this.requestTimeout = requestTimeout;
+        return this;
+    }
+
     @Override
     public String toString() {
         return "DirectConnectionConfig{" +
@@ -198,6 +227,7 @@ public String toString() {
             ", idleEndpointTimeout=" + idleEndpointTimeout +
             ", maxConnectionsPerEndpoint=" + maxConnectionsPerEndpoint +
             ", maxRequestsPerConnection=" + maxRequestsPerConnection +
+            ", requestTimeout=" + requestTimeout +
             '}';
     }
 }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/GatewayConnectionConfig.java

@@ -14,7 +14,7 @@ public final class GatewayConnectionConfig {
     //  Constants
     private static final Duration DEFAULT_REQUEST_TIMEOUT = Duration.ofSeconds(60);
     private static final Duration DEFAULT_IDLE_CONNECTION_TIMEOUT = Duration.ofSeconds(60);
-    private static final int DEFAULT_MAX_POOL_SIZE = 1000;
+    private static final int DEFAULT_MAX_CONNECTION_POOL_SIZE = 1000;
 
     private Duration requestTimeout;
     private int maxConnectionPoolSize;
@@ -26,7 +26,7 @@ public final class GatewayConnectionConfig {
      */
     public GatewayConnectionConfig() {
         this.idleConnectionTimeout = DEFAULT_IDLE_CONNECTION_TIMEOUT;
-        this.maxConnectionPoolSize = DEFAULT_MAX_POOL_SIZE;
+        this.maxConnectionPoolSize = DEFAULT_MAX_CONNECTION_POOL_SIZE;
         this.requestTimeout = DEFAULT_REQUEST_TIMEOUT;
     }
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ConnectionPolicy.java

@@ -18,33 +18,31 @@
  */
 public final class ConnectionPolicy {
 
-    //  Constants
-    public static final Duration DEFAULT_REQUEST_TIMEOUT = Duration.ofSeconds(60);
-    public static final Duration DEFAULT_IDLE_CONNECTION_TIMEOUT = Duration.ofSeconds(60);
-    public static final int DEFAULT_MAX_POOL_SIZE = 1000;
+    private static final int defaultGatewayMaxConnectionPoolSize = GatewayConnectionConfig.getDefaultConfig()
+        .getMaxConnectionPoolSize();
 
-    private static final ConnectionPolicy defaultPolicy = new ConnectionPolicy(DirectConnectionConfig.getDefaultConfig());
+    private static final ConnectionPolicy defaultPolicy =
+        new ConnectionPolicy(DirectConnectionConfig.getDefaultConfig());
 
     private ConnectionMode connectionMode;
-    private String userAgentSuffix;
-    private ThrottlingRetryOptions throttlingRetryOptions;
     private boolean endpointDiscoveryEnabled;
-    private List<String> preferredRegions;
+    private Duration idleConnectionTimeout;
     private boolean multipleWriteRegionsEnabled;
+    private List<String> preferredRegions;
     private boolean readRequestsFallbackEnabled;
+    private ThrottlingRetryOptions throttlingRetryOptions;
+    private String userAgentSuffix;
 
     //  Gateway connection config properties
-    private int maxConnectionPoolSize = DEFAULT_MAX_POOL_SIZE;
-    private Duration requestTimeout = DEFAULT_REQUEST_TIMEOUT;
-    private Duration idleConnectionTimeout = DEFAULT_IDLE_CONNECTION_TIMEOUT;
+    private int maxConnectionPoolSize;
+    private Duration requestTimeout;
     private ProxyOptions proxy;
 
     //  Direct connection config properties
     private Duration connectionTimeout;
-    private Duration idleChannelTimeout;
     private Duration idleEndpointTimeout;
-    private int maxChannelsPerEndpoint;
-    private int maxRequestsPerChannel;
+    private int maxConnectionsPerEndpoint;
+    private int maxRequestsPerConnection;
 
     /**
      * Constructor.
@@ -60,20 +58,22 @@ public ConnectionPolicy(GatewayConnectionConfig gatewayConnectionConfig) {
     public ConnectionPolicy(DirectConnectionConfig directConnectionConfig) {
         this(ConnectionMode.DIRECT);
         this.connectionTimeout = directConnectionConfig.getConnectionTimeout();
-        this.idleChannelTimeout = directConnectionConfig.getIdleConnectionTimeout();
+        this.idleConnectionTimeout = directConnectionConfig.getIdleConnectionTimeout();
         this.idleEndpointTimeout = directConnectionConfig.getIdleEndpointTimeout();
-        this.maxChannelsPerEndpoint = directConnectionConfig.getMaxConnectionsPerEndpoint();
-        this.maxRequestsPerChannel = directConnectionConfig.getMaxRequestsPerConnection();
+        this.maxConnectionsPerEndpoint = directConnectionConfig.getMaxConnectionsPerEndpoint();
+        this.maxRequestsPerConnection = directConnectionConfig.getMaxRequestsPerConnection();
+        this.requestTimeout = directConnectionConfig.getRequestTimeout();
     }
 
     private ConnectionPolicy(ConnectionMode connectionMode) {
         this.connectionMode = connectionMode;
         //  Default values
-        this.throttlingRetryOptions = new ThrottlingRetryOptions();
-        this.userAgentSuffix = "";
-        this.readRequestsFallbackEnabled = true;
         this.endpointDiscoveryEnabled = true;
+        this.maxConnectionPoolSize = defaultGatewayMaxConnectionPoolSize;
         this.multipleWriteRegionsEnabled = true;
+        this.readRequestsFallbackEnabled = true;
+        this.throttlingRetryOptions = new ThrottlingRetryOptions();
+        this.userAgentSuffix = "";
     }
 
     /**
@@ -389,24 +389,6 @@ public ConnectionPolicy setConnectionTimeout(Duration connectionTimeout) {
         return this;
     }
 
-    /**
-     * Gets the idle channel timeout
-     * @return idle channel timeout
-     */
-    public Duration getIdleChannelTimeout() {
-        return idleChannelTimeout;
-    }
-
-    /**
-     * Sets the idle channel timeout
-     * @param idleChannelTimeout idle channel timeout
-     * @return the {@link ConnectionPolicy}
-     */
-    public ConnectionPolicy setIdleChannelTimeout(Duration idleChannelTimeout) {
-        this.idleChannelTimeout = idleChannelTimeout;
-        return this;
-    }
-
     /**
      * Gets the idle endpoint timeout
      * @return the idle endpoint timeout
@@ -429,35 +411,35 @@ public ConnectionPolicy setIdleEndpointTimeout(Duration idleEndpointTimeout) {
      * Gets the max channels per endpoint
      * @return the max channels per endpoint
      */
-    public int getMaxChannelsPerEndpoint() {
-        return maxChannelsPerEndpoint;
+    public int getMaxConnectionsPerEndpoint() {
+        return maxConnectionsPerEndpoint;
     }
 
     /**
      * Sets the max channels per endpoint
-     * @param maxChannelsPerEndpoint the max channels per endpoint
+     * @param maxConnectionsPerEndpoint the max channels per endpoint
      * @return the {@link ConnectionPolicy}
      */
-    public ConnectionPolicy setMaxChannelsPerEndpoint(int maxChannelsPerEndpoint) {
-        this.maxChannelsPerEndpoint = maxChannelsPerEndpoint;
+    public ConnectionPolicy setMaxConnectionsPerEndpoint(int maxConnectionsPerEndpoint) {
+        this.maxConnectionsPerEndpoint = maxConnectionsPerEndpoint;
         return this;
     }
 
     /**
      * Gets the max requests per endpoint
      * @return the max requests per endpoint
      */
-    public int getMaxRequestsPerChannel() {
-        return maxRequestsPerChannel;
+    public int getMaxRequestsPerConnection() {
+        return maxRequestsPerConnection;
     }
 
     /**
      * Sets the max requests per endpoint
-     * @param maxRequestsPerChannel the max requests per endpoint
+     * @param maxRequestsPerConnection the max requests per endpoint
      * @return the {@link ConnectionPolicy}
      */
-    public ConnectionPolicy setMaxRequestsPerChannel(int maxRequestsPerChannel) {
-        this.maxRequestsPerChannel = maxRequestsPerChannel;
+    public ConnectionPolicy setMaxRequestsPerConnection(int maxRequestsPerConnection) {
+        this.maxRequestsPerConnection = maxRequestsPerConnection;
         return this;
     }
 
@@ -477,10 +459,9 @@ public String toString() {
             ", inetSocketProxyAddress=" + proxy.getAddress() +
             ", readRequestsFallbackEnabled=" + readRequestsFallbackEnabled +
             ", connectionTimeout=" + connectionTimeout +
-            ", idleChannelTimeout=" + idleChannelTimeout +
             ", idleEndpointTimeout=" + idleEndpointTimeout +
-            ", maxChannelsPerEndpoint=" + maxChannelsPerEndpoint +
-            ", maxRequestsPerChannel=" + maxRequestsPerChannel +
+            ", maxConnectionsPerEndpoint=" + maxConnectionsPerEndpoint +
+            ", maxRequestsPerConnection=" + maxRequestsPerConnection +
             '}';
     }
 }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/RntbdTransportClient.java

@@ -14,13 +14,13 @@
 import com.azure.cosmos.implementation.directconnectivity.rntbd.RntbdRequestArgs;
 import com.azure.cosmos.implementation.directconnectivity.rntbd.RntbdRequestRecord;
 import com.azure.cosmos.implementation.directconnectivity.rntbd.RntbdServiceEndpoint;
+import com.azure.cosmos.implementation.guava25.base.Strings;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.fasterxml.jackson.databind.ser.std.StdSerializer;
-import com.azure.cosmos.implementation.guava25.base.Strings;
 import io.micrometer.core.instrument.Tag;
 import io.netty.handler.ssl.SslContext;
 import org.slf4j.Logger;
@@ -200,7 +200,6 @@ public static final class Options {
         private final Duration requestExpiryInterval;
 
         @JsonProperty()
-        //  TODO: (DANOBLE) - should we expose it through com.azure.cosmos.DirectConnectionConfig ?
         private final Duration requestTimeout;
 
         @JsonProperty()
@@ -222,26 +221,7 @@ public static final class Options {
 
         // region Constructors
 
-        private Options() {
-            this.bufferPageSize = 8192;
-            this.connectionAcquisitionTimeout = Duration.ZERO;
-            this.connectionTimeout = null;
-            this.idleChannelTimeout = Duration.ZERO;
-            this.idleEndpointTimeout = Duration.ofSeconds(70L);
-            this.maxBufferCapacity = 8192 << 10;
-            this.maxChannelsPerEndpoint = 10;
-            this.maxRequestsPerChannel = 30;
-            this.receiveHangDetectionTime = Duration.ofSeconds(65L);
-            this.requestExpiryInterval = Duration.ofSeconds(5L);
-            this.requestTimeout = null;
-            this.requestTimerResolution = Duration.ofMillis(100L);
-            this.sendHangDetectionTime = Duration.ofSeconds(10L);
-            this.shutdownTimeout = Duration.ofSeconds(15L);
-            this.threadCount = 2 * Runtime.getRuntime().availableProcessors();
-            this.userAgent = new UserAgentContainer();
-        }
-
-        private Options(Builder builder) {
+        private Options(final Builder builder) {
 
             this.bufferPageSize = builder.bufferPageSize;
             this.connectionAcquisitionTimeout = builder.connectionAcquisitionTimeout;
@@ -264,6 +244,25 @@ private Options(Builder builder) {
                 : builder.connectionTimeout;
         }
 
+        private Options(final ConnectionPolicy connectionPolicy) {
+            this.bufferPageSize = 8192;
+            this.connectionAcquisitionTimeout = Duration.ZERO;
+            this.connectionTimeout = connectionPolicy.getConnectionTimeout();
+            this.idleChannelTimeout = connectionPolicy.getIdleConnectionTimeout();
+            this.idleEndpointTimeout = Duration.ofSeconds(70L);
+            this.maxBufferCapacity = 8192 << 10;
+            this.maxChannelsPerEndpoint = connectionPolicy.getMaxConnectionsPerEndpoint();
+            this.maxRequestsPerChannel = connectionPolicy.getMaxRequestsPerConnection();
+            this.receiveHangDetectionTime = Duration.ofSeconds(65L);
+            this.requestExpiryInterval = Duration.ofSeconds(5L);
+            this.requestTimeout = connectionPolicy.getRequestTimeout();
+            this.requestTimerResolution = Duration.ofMillis(100L);
+            this.sendHangDetectionTime = Duration.ofSeconds(10L);
+            this.shutdownTimeout = Duration.ofSeconds(15L);
+            this.threadCount = 2 * Runtime.getRuntime().availableProcessors();
+            this.userAgent = new UserAgentContainer();
+        }
+
         // endregion
 
         // region Accessors
@@ -379,7 +378,8 @@ public String toString() {
          *   "requestTimeout": "PT1M",
          *   "requestTimerResolution": "PT0.5S",
          *   "sendHangDetectionTime": "PT10S",
-         *   "shutdownTimeout": "PT15S"
+         *   "shutdownTimeout": "PT15S",
+         *   "threadCount": 16
          * }}</pre>
          * </li>
          * </ol>
@@ -442,7 +442,7 @@ public static class Builder {
                     }
                 } finally {
                     if (options == null) {
-                        DEFAULT_OPTIONS = new Options();
+                        DEFAULT_OPTIONS = new Options(ConnectionPolicy.getDefaultPolicy());
                     } else {
                         logger.info("Updated default Direct TCP options from system property {}: {}",
                             DEFAULT_OPTIONS_PROPERTY_NAME,
@@ -475,21 +475,17 @@ public static class Builder {
 
             public Builder(ConnectionPolicy connectionPolicy) {
 
-                this.requestTimeout(connectionPolicy.getRequestTimeout());
-
-                //  TODO: (DANOBLE) - Figure out how to get values from connection policy
-                //  At the same time respect the DEFAULT_OPTIONS values - in case user passes some of them.
-
                 this.bufferPageSize = DEFAULT_OPTIONS.bufferPageSize;
                 this.connectionAcquisitionTimeout = DEFAULT_OPTIONS.connectionAcquisitionTimeout;
-                this.connectionTimeout = DEFAULT_OPTIONS.connectionTimeout;
-                this.idleChannelTimeout = DEFAULT_OPTIONS.idleChannelTimeout;
+                this.connectionTimeout = connectionPolicy.getConnectionTimeout();
+                this.idleChannelTimeout = connectionPolicy.getIdleConnectionTimeout();
                 this.idleEndpointTimeout = DEFAULT_OPTIONS.idleEndpointTimeout;
                 this.maxBufferCapacity = DEFAULT_OPTIONS.maxBufferCapacity;
-                this.maxChannelsPerEndpoint = DEFAULT_OPTIONS.maxChannelsPerEndpoint;
-                this.maxRequestsPerChannel = DEFAULT_OPTIONS.maxRequestsPerChannel;
+                this.maxChannelsPerEndpoint = connectionPolicy.getMaxConnectionsPerEndpoint();
+                this.maxRequestsPerChannel = connectionPolicy.getMaxRequestsPerConnection();
                 this.receiveHangDetectionTime = DEFAULT_OPTIONS.receiveHangDetectionTime;
                 this.requestExpiryInterval = DEFAULT_OPTIONS.requestExpiryInterval;
+                this.requestTimeout = connectionPolicy.getRequestTimeout();
                 this.requestTimerResolution = DEFAULT_OPTIONS.requestTimerResolution;
                 this.sendHangDetectionTime = DEFAULT_OPTIONS.sendHangDetectionTime;
                 this.shutdownTimeout = DEFAULT_OPTIONS.shutdownTimeout;