Internal: Allows configuring PendingAcquireMaxCount (Azure#44512)

Internal: Allows configuring PendingAcquireMaxCount for HTTP connection pool in benchmark (to simulate low connection pool size with high concurrency perf workload)

Author: FabianMeiswinkel

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

@@ -63,6 +63,9 @@ public class Configs {
     public static final String HTTP_DEFAULT_CONNECTION_POOL_SIZE = "COSMOS.DEFAULT_HTTP_CONNECTION_POOL_SIZE";
     public static final String HTTP_DEFAULT_CONNECTION_POOL_SIZE_VARIABLE = "COSMOS_DEFAULT_HTTP_CONNECTION_POOL_SIZE";
 
+    public static final String HTTP_PENDING_ACQUIRE_MAX_COUNT = "COSMOS.HTTP_PENDING_ACQUIRE_MAX_COUNT";
+    public static final String HTTP_PENDING_ACQUIRE_MAX_COUNT_VARIABLE = "COSMOS_HTTP_PENDING_ACQUIRE_MAX_COUNT";
+
     public static final String ITEM_SERIALIZATION_INCLUSION_MODE = "COSMOS.ITEM_SERIALIZATION_INCLUSION_MODE";
     public static final String ITEM_SERIALIZATION_INCLUSION_MODE_VARIABLE = "COSMOS_ITEM_SERIALIZATION_INCLUSION_MODE";
 
@@ -520,6 +523,20 @@ public static int getDefaultHttpPoolSize() {
         return DEFAULT_HTTP_DEFAULT_CONNECTION_POOL_SIZE;
     }
 
+    public static Integer getPendingAcquireMaxCount() {
+        String valueFromSystemProperty = System.getProperty(HTTP_PENDING_ACQUIRE_MAX_COUNT);
+        if (valueFromSystemProperty != null && !valueFromSystemProperty.isEmpty()) {
+            return Integer.parseInt(valueFromSystemProperty);
+        }
+
+        String valueFromEnvVariable = System.getenv(HTTP_PENDING_ACQUIRE_MAX_COUNT_VARIABLE);
+        if (valueFromEnvVariable != null && !valueFromEnvVariable.isEmpty()) {
+            return Integer.parseInt(valueFromEnvVariable);
+        }
+
+        return null;
+    }
+
     private static String validateSerializationInclusionMode(String serializationInclusionMode) {
         if (!Strings.isNullOrEmpty(serializationInclusionMode)) {
             if ("Always".equalsIgnoreCase(serializationInclusionMode)

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

@@ -55,6 +55,8 @@ public final class ConnectionPolicy {
     private int aggressiveWarmupConcurrency;
     private boolean serverCertValidationDisabled = false;
 
+    private Integer pendingAcquireMaxCount;
+
     /**
      * Constructor.
      */
@@ -108,6 +110,8 @@ private ConnectionPolicy(
                     .DirectConnectionConfigHelper
                     .getDirectConnectionConfigAccessor()
                     .getMinConnectionPoolSizePerEndpoint(directConnectionConfig), Configs.getMinConnectionPoolSizePerEndpoint());
+
+        this.pendingAcquireMaxCount = Configs.getPendingAcquireMaxCount();
     }
 
     private ConnectionPolicy() {
@@ -122,6 +126,7 @@ private ConnectionPolicy() {
         this.minConnectionPoolSizePerEndpoint = Configs.getMinConnectionPoolSizePerEndpoint();
         this.openConnectionsConcurrency = Configs.getOpenConnectionsConcurrency();
         this.aggressiveWarmupConcurrency = Configs.getAggressiveWarmupConcurrency();
+        this.pendingAcquireMaxCount = Configs.getPendingAcquireMaxCount();
     }
 
     /**
@@ -637,6 +642,7 @@ public boolean isServerCertValidationDisabled() {
 
     @Override
     public String toString() {
+
         return "ConnectionPolicy{" +
             "httpNetworkRequestTimeout=" + httpNetworkRequestTimeout +
             ", tcpNetworkRequestTimeout=" + tcpNetworkRequestTimeout +
@@ -663,6 +669,7 @@ public String toString() {
             ", minConnectionPoolSizePerEndpoint=" + minConnectionPoolSizePerEndpoint +
             ", openConnectionsConcurrency=" + openConnectionsConcurrency +
             ", aggressiveWarmupConcurrency=" + aggressiveWarmupConcurrency +
+            ", pendingAcquireMaxCount=" + this.pendingAcquireMaxCount != null ? String.valueOf(this.pendingAcquireMaxCount) : "DEFAULT" +
             '}';
     }
 }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/HttpClient.java

@@ -42,7 +42,12 @@ static HttpClient createFixed(HttpClientConfig httpClientConfig) {
 
         ConnectionProvider.Builder fixedConnectionProviderBuilder = ConnectionProvider
             .builder(httpClientConfig.getConnectionPoolName());
+
         fixedConnectionProviderBuilder.maxConnections(httpClientConfig.getMaxPoolSize());
+        Integer customPendingAcquireMaxCount = httpClientConfig.getPendingAcquireMaxCount();
+        if (customPendingAcquireMaxCount != null) {
+            fixedConnectionProviderBuilder.pendingAcquireMaxCount(customPendingAcquireMaxCount);
+        }
         fixedConnectionProviderBuilder.pendingAcquireTimeout(httpClientConfig.getConnectionAcquireTimeout());
         fixedConnectionProviderBuilder.maxIdleTime(httpClientConfig.getMaxIdleConnectionTimeout());
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/HttpClientConfig.java

@@ -16,6 +16,8 @@ public class HttpClientConfig {
     private final Configs configs;
     private Duration connectionAcquireTimeout = Configs.getConnectionAcquireTimeout();
     private int maxPoolSize = Configs.getDefaultHttpPoolSize();
+
+    private Integer pendingAcquireMaxCount = Configs.getPendingAcquireMaxCount();
     private Duration maxIdleConnectionTimeout = Configs.getMaxIdleConnectionTimeout();
     private Duration networkRequestTimeout = Duration.ofSeconds(Configs.getHttpResponseTimeoutInSeconds());
     private String connectionPoolName = Configs.getReactorNettyConnectionPoolName();
@@ -112,6 +114,10 @@ public int getMaxPoolSize() {
         return maxPoolSize;
     }
 
+    public Integer getPendingAcquireMaxCount() {
+        return pendingAcquireMaxCount;
+    }
+
     public Duration getMaxIdleConnectionTimeout() {
         return maxIdleConnectionTimeout;
     }