fix: Fix channel clamping logic in ChannelPools to respect channel bounds

Author: lqiu96

sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/ChannelPool.java

@@ -310,6 +310,10 @@ void resize() {
     if (minChannels < settings.getMinChannelCount()) {
       minChannels = settings.getMinChannelCount();
     }
+    // Limit in case the calculated min channel count exceeds the configured max channel count
+    if (minChannels > settings.getMaxChannelCount()) {
+      minChannels = settings.getMaxChannelCount();
+    }
 
     // Number of channels if each channel operated at minimum capacity
     // Note: getMinRpcsPerChannel() can return 0, but division by 0 shouldn't cause a problem.
@@ -319,8 +323,9 @@ void resize() {
     if (maxChannels > settings.getMaxChannelCount()) {
       maxChannels = settings.getMaxChannelCount();
     }
-    if (maxChannels < minChannels) {
-      maxChannels = minChannels;
+    // Limit in case the calculated max channel count falls below the configured min channel count
+    if (maxChannels < settings.getMinChannelCount()) {
+      maxChannels = settings.getMinChannelCount();
     }
 
     // If the pool were to be resized, try to aim for the middle of the bound, but limit rate of

sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/ChannelPoolSettings.java

@@ -182,7 +182,7 @@ public ChannelPoolSettings build() {
       Preconditions.checkState(
           s.getMinChannelCount() > 0, "Minimum channel count must be at least 1");
       Preconditions.checkState(
-          s.getMinChannelCount() <= s.getMaxRpcsPerChannel(), "absolute channel range is invalid");
+          s.getMinChannelCount() <= s.getMaxChannelCount(), "absolute channel range is invalid");
       Preconditions.checkState(
           s.getMinChannelCount() <= s.getInitialChannelCount(),
           "initial channel count be at least minChannelCount");

sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/ChannelPoolTest.java

@@ -144,6 +144,26 @@ private void verifyTargetChannel(
     }
   }
 
+  private static ChannelFactory createMockChannelFactory(
+      List<ManagedChannel> channels, List<ClientCall<Object, Object>> startedCalls) {
+    return () -> {
+      ManagedChannel channel = Mockito.mock(ManagedChannel.class);
+      Mockito.when(channel.newCall(Mockito.any(), Mockito.any()))
+          .thenAnswer(
+              invocation -> {
+                @SuppressWarnings("unchecked")
+                ClientCall<Object, Object> clientCall = Mockito.mock(ClientCall.class);
+                if (startedCalls != null) {
+                  startedCalls.add(clientCall);
+                }
+                return clientCall;
+              });
+
+      channels.add(channel);
+      return channel;
+    };
+  }
+
   @Test
   void ensureEvenDistribution() throws InterruptedException, IOException {
     int numChannels = 10;
@@ -451,21 +471,7 @@ void channelCountShouldNotChangeWhenOutstandingRpcsAreWithinLimits() throws Exce
     List<ManagedChannel> channels = new ArrayList<>();
     List<ClientCall<Object, Object>> startedCalls = new ArrayList<>();
 
-    ChannelFactory channelFactory =
-        () -> {
-          ManagedChannel channel = Mockito.mock(ManagedChannel.class);
-          Mockito.when(channel.newCall(Mockito.any(), Mockito.any()))
-              .thenAnswer(
-                  invocation -> {
-                    @SuppressWarnings("unchecked")
-                    ClientCall<Object, Object> clientCall = Mockito.mock(ClientCall.class);
-                    startedCalls.add(clientCall);
-                    return clientCall;
-                  });
-
-          channels.add(channel);
-          return channel;
-        };
+    ChannelFactory channelFactory = createMockChannelFactory(channels, startedCalls);
 
     pool =
         new ChannelPool(
@@ -531,21 +537,7 @@ void customResizeDeltaIsRespected() throws Exception {
     List<ManagedChannel> channels = new ArrayList<>();
     List<ClientCall<Object, Object>> startedCalls = new ArrayList<>();
 
-    ChannelFactory channelFactory =
-        () -> {
-          ManagedChannel channel = Mockito.mock(ManagedChannel.class);
-          Mockito.when(channel.newCall(Mockito.any(), Mockito.any()))
-              .thenAnswer(
-                  invocation -> {
-                    @SuppressWarnings("unchecked")
-                    ClientCall<Object, Object> clientCall = Mockito.mock(ClientCall.class);
-                    startedCalls.add(clientCall);
-                    return clientCall;
-                  });
-
-          channels.add(channel);
-          return channel;
-        };
+    ChannelFactory channelFactory = createMockChannelFactory(channels, startedCalls);
 
     pool =
         new ChannelPool(
@@ -578,21 +570,7 @@ void removedIdleChannelsAreShutdown() throws Exception {
     List<ManagedChannel> channels = new ArrayList<>();
     List<ClientCall<Object, Object>> startedCalls = new ArrayList<>();
 
-    ChannelFactory channelFactory =
-        () -> {
-          ManagedChannel channel = Mockito.mock(ManagedChannel.class);
-          Mockito.when(channel.newCall(Mockito.any(), Mockito.any()))
-              .thenAnswer(
-                  invocation -> {
-                    @SuppressWarnings("unchecked")
-                    ClientCall<Object, Object> clientCall = Mockito.mock(ClientCall.class);
-                    startedCalls.add(clientCall);
-                    return clientCall;
-                  });
-
-          channels.add(channel);
-          return channel;
-        };
+    ChannelFactory channelFactory = createMockChannelFactory(channels, startedCalls);
 
     pool =
         new ChannelPool(
@@ -619,21 +597,7 @@ void removedActiveChannelsAreShutdown() throws Exception {
     List<ManagedChannel> channels = new ArrayList<>();
     List<ClientCall<Object, Object>> startedCalls = new ArrayList<>();
 
-    ChannelFactory channelFactory =
-        () -> {
-          ManagedChannel channel = Mockito.mock(ManagedChannel.class);
-          Mockito.when(channel.newCall(Mockito.any(), Mockito.any()))
-              .thenAnswer(
-                  invocation -> {
-                    @SuppressWarnings("unchecked")
-                    ClientCall<Object, Object> clientCall = Mockito.mock(ClientCall.class);
-                    startedCalls.add(clientCall);
-                    return clientCall;
-                  });
-
-          channels.add(channel);
-          return channel;
-        };
+    ChannelFactory channelFactory = createMockChannelFactory(channels, startedCalls);
 
     pool =
         new ChannelPool(
@@ -734,21 +698,7 @@ void repeatedResizingLogsWarningOnExpand() throws Exception {
     List<ManagedChannel> channels = new ArrayList<>();
     List<ClientCall<Object, Object>> startedCalls = new ArrayList<>();
 
-    ChannelFactory channelFactory =
-        () -> {
-          ManagedChannel channel = Mockito.mock(ManagedChannel.class);
-          Mockito.when(channel.newCall(Mockito.any(), Mockito.any()))
-              .thenAnswer(
-                  invocation -> {
-                    @SuppressWarnings("unchecked")
-                    ClientCall<Object, Object> clientCall = Mockito.mock(ClientCall.class);
-                    startedCalls.add(clientCall);
-                    return clientCall;
-                  });
-
-          channels.add(channel);
-          return channel;
-        };
+    ChannelFactory channelFactory = createMockChannelFactory(channels, startedCalls);
 
     pool =
         new ChannelPool(
@@ -899,4 +849,75 @@ void testDoubleRelease() throws Exception {
       ChannelPool.LOG.removeHandler(logHandler);
     }
   }
+
+  @Test
+  void settingsValidationFailsWhenMinChannelsExceedsMaxChannels() {
+    Assertions.assertThrows(
+        IllegalStateException.class,
+        () -> ChannelPoolSettings.builder().setMinChannelCount(2).setMaxChannelCount(1).build());
+  }
+
+  @Test
+  void minChannelsClampedToMaxChannelCountUnderHighLoad() throws Exception {
+    ScheduledExecutorService executor = Mockito.mock(ScheduledExecutorService.class);
+    FixedExecutorProvider provider = FixedExecutorProvider.create(executor);
+
+    List<ManagedChannel> channels = new ArrayList<>();
+    ChannelFactory channelFactory = createMockChannelFactory(channels, null);
+
+    pool =
+        new ChannelPool(
+            ChannelPoolSettings.builder()
+                .setInitialChannelCount(1)
+                .setMinRpcsPerChannel(1)
+                .setMaxRpcsPerChannel(2)
+                .setMaxResizeDelta(10) // Allow large growth
+                .setMinChannelCount(1)
+                .setMaxChannelCount(5)
+                .build(),
+            channelFactory,
+            provider);
+    assertThat(pool.entries.get()).hasSize(1);
+
+    // Add 20 RPCs, which would require 10 channels (20/2)
+    // But max is 5
+    for (int i = 0; i < 20; i++) {
+      ClientCalls.futureUnaryCall(
+          pool.newCall(METHOD_RECOGNIZE, CallOptions.DEFAULT), Color.getDefaultInstance());
+    }
+
+    pool.resize();
+
+    // Should be clamped to maxChannelCount = 5
+    assertThat(pool.entries.get()).hasSize(5);
+  }
+
+  @Test
+  void maxChannelsClampedToMinChannelCountUnderLowLoad() throws Exception {
+    ScheduledExecutorService executor = Mockito.mock(ScheduledExecutorService.class);
+    FixedExecutorProvider provider = FixedExecutorProvider.create(executor);
+
+    List<ManagedChannel> channels = new ArrayList<>();
+    ChannelFactory channelFactory = createMockChannelFactory(channels, null);
+
+    pool =
+        new ChannelPool(
+            ChannelPoolSettings.builder()
+                .setInitialChannelCount(5)
+                .setMinRpcsPerChannel(1)
+                .setMaxRpcsPerChannel(2)
+                .setMinChannelCount(3)
+                .setMaxChannelCount(10)
+                .build(),
+            channelFactory,
+            provider);
+    assertThat(pool.entries.get()).hasSize(5);
+
+    // With no outstanding RPCs, the pool should want to shrink to 0
+    // But min is 3
+    pool.resize();
+
+    // Should be clamped to minChannelCount = 3
+    assertThat(pool.entries.get()).hasSize(3);
+  }
 }