test: added connection leakage test for OkHttp

Signed-off-by: Marc Nuri <marc@marcnuri.com>

CHANGELOG.md

@@ -3,7 +3,6 @@
 ### 6.4-SNAPSHOT
 
 #### Bugs
-* Fix #4666: fixed okhttp calls not explicitly closing
 
 #### Improvements
 
@@ -13,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

@@ -75,13 +75,18 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>io.fabric8</groupId>
-      <artifactId>mockwebserver</artifactId>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-params</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
-      <artifactId>mockito-inline</artifactId>
+      <artifactId>mockito-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.fabric8</groupId>
+      <artifactId>mockwebserver</artifactId>
+      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.assertj</groupId>

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

@@ -87,7 +87,7 @@ abstract static class OkHttpAsyncBody<T> implements AsyncBody {
     private final CompletableFuture<Void> done = new CompletableFuture<>();
     private boolean consuming;
     private boolean requested;
-    private Executor executor;
+    private final Executor executor;
 
     OkHttpAsyncBody(AsyncBody.Consumer<T> consumer, BufferedSource source, Executor executor) {
       this.consumer = consumer;

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

@@ -17,14 +17,6 @@
 
 import io.fabric8.kubernetes.client.http.AbstractAsyncBodyTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
-import io.fabric8.kubernetes.client.okhttp.OkHttpClientImpl.OkHttpAsyncBody;
-import okio.BufferedSource;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.List;
 
 @SuppressWarnings("java:S2187")
 public class OkHttpAsyncBodyTest extends AbstractAsyncBodyTest {
@@ -33,20 +25,4 @@ protected HttpClient.Factory getHttpClientFactory() {
     return new OkHttpClientFactory();
   }
 
-  @Test
-  void testClosedWhenExhausted() throws Exception {
-    BufferedSource source = Mockito.mock(BufferedSource.class);
-    Mockito.when(source.exhausted()).thenReturn(true);
-    OkHttpClientImpl.OkHttpAsyncBody<List<ByteBuffer>> asyncBody = new OkHttpAsyncBody<List<ByteBuffer>>(null, source,
-        Runnable::run) {
-
-      @Override
-      protected List<ByteBuffer> process(BufferedSource source) throws IOException {
-        return null;
-      }
-    };
-
-    asyncBody.consume();
-    Mockito.verify(source).close();
-  }
 }

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();
+  }
+}