fix #4666: addressing okhttp sources needing explicit close

CHANGELOG.md

@@ -12,6 +12,11 @@
 
 #### _**Note**_: Breaking changes
 
+### 6.3.1-SNAPSHOT
+
+#### Bugs
+* Fix #4666: fixed okhttp calls not explicitly closing
+
 ### 6.3.0 (2022-12-12)
 
 #### Bugs

httpclient-okhttp/pom.xml

@@ -74,6 +74,15 @@
       <artifactId>junit-jupiter-engine</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-params</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.fabric8</groupId>
       <artifactId>mockwebserver</artifactId>

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

@@ -62,6 +62,7 @@
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.function.Function;
@@ -80,16 +81,18 @@ static MediaType parseMediaType(String contentType) {
     return result;
   }
 
-  private abstract class OkHttpAsyncBody<T> implements AsyncBody {
+  abstract static class OkHttpAsyncBody<T> implements AsyncBody {
     private final AsyncBody.Consumer<T> consumer;
     private final BufferedSource source;
     private final CompletableFuture<Void> done = new CompletableFuture<>();
     private boolean consuming;
     private boolean requested;
+    private final Executor executor;
 
-    private OkHttpAsyncBody(AsyncBody.Consumer<T> consumer, BufferedSource source) {
+    OkHttpAsyncBody(AsyncBody.Consumer<T> consumer, BufferedSource source, Executor executor) {
       this.consumer = consumer;
       this.source = source;
+      this.executor = executor;
     }
 
     @Override
@@ -103,7 +106,7 @@ public void consume() {
       }
       try {
         // consume should not block a caller, delegate to the dispatcher thread pool
-        httpClient.dispatcher().executorService().execute(this::doConsume);
+        executor.execute(this::doConsume);
       } catch (Exception e) {
         // executor is likely shutdown
         Utils.closeQuietly(source);
@@ -125,6 +128,8 @@ private void doConsume() {
             T value = process(source);
             consumer.consume(value, this);
           } else {
+            // even if we've read everything an explicit close is still needed
+            source.close();
             done.complete(null);
           }
         }
@@ -311,7 +316,8 @@ private okhttp3.Request.Builder newRequestBuilder() {
   @Override
   public CompletableFuture<HttpResponse<AsyncBody>> consumeBytesDirect(StandardHttpRequest request,
       Consumer<List<ByteBuffer>> consumer) {
-    Function<BufferedSource, AsyncBody> handler = s -> new OkHttpAsyncBody<List<ByteBuffer>>(consumer, s) {
+    Function<BufferedSource, AsyncBody> handler = s -> new OkHttpAsyncBody<List<ByteBuffer>>(consumer, s,
+        this.httpClient.dispatcher().executorService()) {
       @Override
       protected List<ByteBuffer> process(BufferedSource source) throws IOException {
         // read only what is available otherwise okhttp will block trying to read

httpclient-okhttp/src/test/java/io/fabric8/kubernetes/client/okhttp/ConnectionPoolLeakageTest.java

@@ -0,0 +1,86 @@
+/**
+ * Copyright (C) 2015 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.fabric8.kubernetes.client.okhttp;
+
+import io.fabric8.kubernetes.client.http.AsyncBody;
+import io.fabric8.kubernetes.client.http.HttpClient;
+import io.fabric8.kubernetes.client.http.HttpResponse;
+import okhttp3.ConnectionPool;
+import okhttp3.Protocol;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ConnectionPoolLeakageTest {
+
+  private MockWebServer server;
+
+  private ConnectionPool connectionPool;
+  private OkHttpClientBuilderImpl clientBuilder;
+
+  @BeforeEach
+  void setUp() {
+    server = new MockWebServer();
+    final char[] chars = new char[10485760];
+    Arrays.fill(chars, '1');
+    server.enqueue(new MockResponse().setResponseCode(200).setChunkedBody(new String(chars), 1024));
+    connectionPool = new ConnectionPool(10, 100, TimeUnit.SECONDS);
+    clientBuilder = new OkHttpClientFactory().newBuilder();
+    clientBuilder.getBuilder().connectionPool(connectionPool);
+  }
+
+  @AfterEach
+  void tearDown() throws Exception {
+    server.shutdown();
+    connectionPool.evictAll();
+  }
+
+  @ParameterizedTest(name = "with protocol {0}")
+  @DisplayName("consumeBytes should not leak connections")
+  @ValueSource(strings = { "h2_prior_knowledge", "http/1.1" })
+  void consumeBytes(String protocol) throws Exception {
+    final Protocol p = Protocol.get(protocol);
+    server.setProtocols(Collections.singletonList(p));
+    server.start();
+    clientBuilder.getBuilder().protocols(Collections.singletonList(p));
+    try (HttpClient httpClient = clientBuilder.build()) {
+      final HttpResponse<AsyncBody> asyncBodyResponse = httpClient.consumeBytes(
+          httpClient.newHttpRequestBuilder().uri(server.url("/").toString()).build(),
+          (value, asyncBody) -> {
+            asyncBody.consume();
+          })
+          .get(10L, TimeUnit.SECONDS);
+      assertThat(activeConnections()).isEqualTo(1);
+      asyncBodyResponse.body().consume();
+      asyncBodyResponse.body().done().get(10L, TimeUnit.SECONDS);
+      assertThat(activeConnections()).isZero();
+    }
+  }
+
+  private int activeConnections() {
+    return connectionPool.connectionCount() - connectionPool.idleConnectionCount();
+  }
+}

httpclient-okhttp/src/test/java/io/fabric8/kubernetes/client/okhttp/OkHttpAsyncBodyTest.java

@@ -24,4 +24,5 @@ public class OkHttpAsyncBodyTest extends AbstractAsyncBodyTest {
   protected HttpClient.Factory getHttpClientFactory() {
     return new OkHttpClientFactory();
   }
+
 }

httpclient-okhttp/src/test/java/io/fabric8/kubernetes/client/okhttp/OkHttpImplAsyncBodyTest.java

@@ -0,0 +1,47 @@
+/**
+ * Copyright (C) 2015 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.fabric8.kubernetes.client.okhttp;
+
+import okio.BufferedSource;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class OkHttpImplAsyncBodyTest {
+
+  @Test
+  void testClosedWhenExhausted() throws Exception {
+    BufferedSource source = mock(BufferedSource.class);
+    when(source.exhausted()).thenReturn(true);
+    OkHttpClientImpl.OkHttpAsyncBody<List<ByteBuffer>> asyncBody = new OkHttpClientImpl.OkHttpAsyncBody<List<ByteBuffer>>(null,
+        source,
+        Runnable::run) {
+
+      @Override
+      protected List<ByteBuffer> process(BufferedSource source) {
+        return null;
+      }
+    };
+
+    asyncBody.consume();
+    Mockito.verify(source).close();
+  }
+}