fix fabric8io#4666: addressing okhttp sources needing explicit close

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 6.4-SNAPSHOT
 
 #### Bugs
+* Fix #4666: fixed okhttp calls not explicitly closing
 
 #### Improvements
 

httpclient-okhttp/pom.xml

@@ -79,6 +79,10 @@
       <artifactId>mockwebserver</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-inline</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.assertj</groupId>
       <artifactId>assertj-core</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 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/OkHttpAsyncBodyTest.java

@@ -17,11 +17,36 @@
 
 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 {
   @Override
   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();
+  }
 }