fix fabric8io#4663: changes to help prevent the client from waiting i…

…ndefinitely

CHANGELOG.md

@@ -11,6 +11,7 @@
 #### New Features
 
 #### _**Note**_: Breaking changes
+* Fix #4663: Config.maxConcurrentRequests and Config.maxConcurrentRequestsPerHost will no longer be used.  Instead they will default to unlimited for all clients.  Due to the ability of the fabric8 client to start long running requests (either websocket or regular http) and how this is treated by the underlying clients you can easily exhaust these values and enter a state where the client is unresponsive without any additional information on what is occurring.
 
 ### 6.3.0 (2022-12-12)
 

httpclient-jetty/src/main/java/io/fabric8/kubernetes/client/jetty/JettyHttpClientBuilder.java

@@ -58,6 +58,9 @@ public JettyHttpClient build() {
       sslContextFactory.setIncludeProtocols(Stream.of(tlsVersions).map(TlsVersion::javaName).toArray(String[]::new));
     }
     HttpClient sharedHttpClient = new HttpClient(newTransport(sslContextFactory, preferHttp11));
+    // long running http requests count against this and eventually exhaust
+    // the work that can be done
+    sharedHttpClient.setMaxConnectionsPerDestination(Integer.MAX_VALUE);
     WebSocketClient sharedWebSocketClient = new WebSocketClient(new HttpClient(newTransport(sslContextFactory, preferHttp11)));
     sharedWebSocketClient.setIdleTimeout(Duration.ZERO);
     if (connectTimeout != null) {

httpclient-okhttp/src/main/java/io/fabric8/kubernetes/client/okhttp/OkHttpClientFactory.java

@@ -84,12 +84,14 @@ public OkHttpClientBuilderImpl newBuilder(Config config) {
         httpClientBuilder.pingInterval(config.getWebsocketPingInterval(), TimeUnit.MILLISECONDS);
       }
 
-      if (config.getMaxConcurrentRequests() > 0 && config.getMaxConcurrentRequestsPerHost() > 0) {
-        Dispatcher dispatcher = new Dispatcher();
-        dispatcher.setMaxRequests(config.getMaxConcurrentRequests());
-        dispatcher.setMaxRequestsPerHost(config.getMaxConcurrentRequestsPerHost());
-        httpClientBuilder.dispatcher(dispatcher);
-      }
+      Dispatcher dispatcher = new Dispatcher();
+      // websockets and long running http requests count against this and eventually starve
+      // the work that can be done
+      dispatcher.setMaxRequests(Integer.MAX_VALUE);
+      // long running http requests count against this and eventually exhaust
+      // the work that can be done
+      dispatcher.setMaxRequestsPerHost(Integer.MAX_VALUE);
+      httpClientBuilder.dispatcher(dispatcher);
 
       HttpClientUtils.applyCommonConfiguration(config, builderWrapper, this);
 

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/OperationSupport.java

@@ -60,9 +60,11 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 public class OperationSupport {
 
+  private static final long ADDITIONAL_REQEUST_TIMEOUT = TimeUnit.SECONDS.toMillis(5);
   private static final String FIELD_MANAGER_PARAM = "?fieldManager=";
   public static final String JSON = "application/json";
   public static final String JSON_PATCH = "application/json-patch+json";
@@ -516,7 +518,11 @@ protected <T> T handleRawGet(URL resourceUrl, Class<T> type) throws IOException
    */
   protected <T> T waitForResult(CompletableFuture<T> future) throws IOException {
     try {
-      // readTimeout should be enforced
+      // since readTimeout may not be enforced in a timely manner at the httpclient, we'll
+      // enforce a higher level timeout with a small amount of padding to account for possible queuing
+      if (config.getRequestTimeout() > 0) {
+        return future.get(config.getRequestTimeout() + ADDITIONAL_REQEUST_TIMEOUT, TimeUnit.MILLISECONDS);
+      }
       return future.get();
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
@@ -536,6 +542,9 @@ protected <T> T waitForResult(CompletableFuture<T> future) throws IOException {
         throw ((KubernetesClientException) t).copyAsCause();
       }
       throw new KubernetesClientException(t.getMessage(), t);
+    } catch (TimeoutException e) {
+      future.cancel(true);
+      throw KubernetesClientException.launderThrowable(e);
     }
   }
 

kubernetes-client/src/test/java/io/fabric8/kubernetes/client/okhttp/OkHttpClientFactoryTest.java

@@ -25,32 +25,32 @@
 class OkHttpClientFactoryTest {
 
   @Test
-  void shouldRespectMaxRequests() {
+  void shouldNotRespectMaxRequests() {
     OkHttpClientImpl client = new OkHttpClientFactory().newBuilder(new ConfigBuilder().build()).build();
 
-    assertEquals(64, client.getOkHttpClient().dispatcher().getMaxRequests());
+    assertEquals(Integer.MAX_VALUE, client.getOkHttpClient().dispatcher().getMaxRequests());
 
     Config config = new ConfigBuilder()
         .withMaxConcurrentRequests(120)
         .build();
 
     client = new OkHttpClientFactory().newBuilder(config).build();
-    assertEquals(120, client.getOkHttpClient().dispatcher().getMaxRequests());
+    assertEquals(Integer.MAX_VALUE, client.getOkHttpClient().dispatcher().getMaxRequests());
   }
 
   @Test
-  void shouldRespectMaxRequestsPerHost() {
+  void shouldNotRespectMaxRequestsPerHost() {
     OkHttpClientImpl client = new OkHttpClientFactory().newBuilder(new ConfigBuilder().build()).build();
 
-    assertEquals(5, client.getOkHttpClient().dispatcher().getMaxRequestsPerHost());
+    assertEquals(Integer.MAX_VALUE, client.getOkHttpClient().dispatcher().getMaxRequestsPerHost());
 
     Config config = new ConfigBuilder()
         .withMaxConcurrentRequestsPerHost(20)
         .build();
 
     client = new OkHttpClientFactory().newBuilder(config).build();
 
-    assertEquals(20, client.getOkHttpClient().dispatcher().getMaxRequestsPerHost());
+    assertEquals(Integer.MAX_VALUE, client.getOkHttpClient().dispatcher().getMaxRequestsPerHost());
   }
 
 }