fix #4201: ensuring okhttp is tolerant to multiple consume calls

also converting jetty to fully non-blocking

and adapting the body delivery for jdk for use with sendAsync logic - we
need the AsyncBody to be available immediately

Author: shawkins

httpclient-jdk/src/main/java/io/fabric8/kubernetes/client/jdkhttp/JdkHttpClientImpl.java

@@ -26,17 +26,19 @@
 import java.net.URI;
 import java.net.http.HttpResponse.BodyHandler;
 import java.net.http.HttpResponse.BodyHandlers;
+import java.net.http.HttpResponse.BodySubscriber;
+import java.net.http.HttpResponse.ResponseInfo;
 import java.net.http.WebSocketHandshakeException;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Subscriber;
 import java.util.concurrent.Flow.Subscription;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 
@@ -47,48 +49,80 @@
  */
 public class JdkHttpClientImpl implements HttpClient {
 
+  /**
+   * Adapts the BodyHandler to immediately complete the body
+   */
+  private static final class BodyHandlerAdapter implements BodyHandler<AsyncBody> {
+    private final AsyncBodySubscriber<?> subscriber;
+    private final BodyHandler<Void> handler;
+
+    private BodyHandlerAdapter(AsyncBodySubscriber<?> subscriber, BodyHandler<Void> handler) {
+      this.subscriber = subscriber;
+      this.handler = handler;
+    }
+
+    @Override
+    public BodySubscriber<AsyncBody> apply(ResponseInfo responseInfo) {
+      BodySubscriber<Void> bodySubscriber = handler.apply(responseInfo);
+      return new BodySubscriber<AsyncBody>() {
+        CompletableFuture<AsyncBody> cf = CompletableFuture.completedFuture(subscriber);
+
+        @Override
+        public void onSubscribe(Subscription subscription) {
+          bodySubscriber.onSubscribe(subscription);
+        }
+
+        @Override
+        public void onNext(List<ByteBuffer> item) {
+          bodySubscriber.onNext(item);
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+          bodySubscriber.onError(throwable);
+        }
+
+        @Override
+        public void onComplete() {
+          bodySubscriber.onComplete();
+        }
+
+        @Override
+        public CompletionStage<AsyncBody> getBody() {
+          return cf;
+        }
+      };
+    }
+  }
+
   private final class AsyncBodySubscriber<T> implements Subscriber<T>, AsyncBody {
     private final BodyConsumer<T> consumer;
     private CompletableFuture<Void> done = new CompletableFuture<Void>();
-    private final AtomicBoolean subscribed = new AtomicBoolean();
-    private volatile Flow.Subscription subscription;
-    private T initialItem;
-    private boolean first = true;
-    private boolean isComplete;
+    private CompletableFuture<Flow.Subscription> subscription = new CompletableFuture<>();
 
     private AsyncBodySubscriber(BodyConsumer<T> consumer) {
       this.consumer = consumer;
     }
 
     @Override
     public void onSubscribe(Subscription subscription) {
-      if (!subscribed.compareAndSet(false, true)) {
+      if (this.subscription.isDone()) {
         subscription.cancel();
         return;
       }
-      this.subscription = subscription;
-      // the sendAsync future won't complete unless we do the initial request here
-      // so in onNext we'll trap the item until we're ready
-      subscription.request(1);
+      this.subscription.complete(subscription);
     }
 
     @Override
     public void onNext(T item) {
-      synchronized (this) {
-        if (first) {
-          this.initialItem = item;
-          first = false;
-          return;
-        }
-      }
       try {
         if (item == null) {
           done.complete(null);
         } else {
           consumer.consume(item, this);
         }
       } catch (Exception e) {
-        subscription.cancel();
+        subscription.thenAccept(Subscription::cancel);
         done.completeExceptionally(e);
       }
     }
@@ -100,10 +134,6 @@ public void onError(Throwable throwable) {
 
     @Override
     public synchronized void onComplete() {
-      if (initialItem != null) {
-        this.isComplete = true;
-        return;
-      }
       done.complete(null);
     }
 
@@ -112,19 +142,7 @@ public synchronized void consume() {
       if (done.isDone()) {
         return;
       }
-      try {
-        first = false;
-        if (initialItem != null) {
-          T item = initialItem;
-          initialItem = null;
-          onNext(item);
-        }
-      } finally {
-        if (isComplete) {
-          done.complete(null);
-        }
-        this.subscription.request(1);
-      }
+      this.subscription.thenAccept(s -> s.request(1));
     }
 
     @Override
@@ -134,7 +152,7 @@ public CompletableFuture<Void> done() {
 
     @Override
     public void cancel() {
-      subscription.cancel();
+      subscription.thenAccept(Subscription::cancel);
       done.cancel(false);
     }
 
@@ -234,7 +252,8 @@ public CompletableFuture<HttpResponse<AsyncBody>> consumeLines(HttpRequest reque
     return sendAsync(request, () -> {
       AsyncBodySubscriber<String> subscriber = new AsyncBodySubscriber<>(consumer);
       BodyHandler<Void> handler = BodyHandlers.fromLineSubscriber(subscriber);
-      return new HandlerAndAsyncBody<>(handler, subscriber);
+      BodyHandler<AsyncBody> handlerAdapter = new BodyHandlerAdapter(subscriber, handler);
+      return new HandlerAndAsyncBody<>(handlerAdapter, subscriber);
     }).thenApply(r -> new JdkHttpResponseImpl<AsyncBody>(r.response, r.asyncBody));
   }
 
@@ -243,7 +262,8 @@ public CompletableFuture<HttpResponse<AsyncBody>> consumeBytes(HttpRequest reque
     return sendAsync(request, () -> {
       AsyncBodySubscriber<List<ByteBuffer>> subscriber = new AsyncBodySubscriber<>(consumer);
       BodyHandler<Void> handler = BodyHandlers.fromSubscriber(subscriber);
-      return new HandlerAndAsyncBody<>(handler, subscriber);
+      BodyHandler<AsyncBody> handlerAdapter = new BodyHandlerAdapter(subscriber, handler);
+      return new HandlerAndAsyncBody<>(handlerAdapter, subscriber);
     }).thenApply(r -> new JdkHttpResponseImpl<AsyncBody>(r.response, r.asyncBody));
   }
 

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

@@ -15,37 +15,47 @@
  */
 package io.fabric8.kubernetes.client.jetty;
 
-import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.http.HttpClient;
 import io.fabric8.kubernetes.client.http.HttpRequest;
 import io.fabric8.kubernetes.client.http.HttpResponse;
 import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.client.api.Response;
 import org.eclipse.jetty.client.api.Result;
+import org.eclipse.jetty.util.Callback;
 
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.LongConsumer;
 
 public abstract class JettyAsyncResponseListener<T> extends Response.Listener.Adapter implements HttpClient.AsyncBody {
 
   private final HttpRequest httpRequest;
   private final HttpClient.BodyConsumer<T> bodyConsumer;
   private final CompletableFuture<HttpResponse<HttpClient.AsyncBody>> asyncResponse;
   private final CompletableFuture<Void> asyncBodyDone;
-  private boolean consume;
+  private volatile LongConsumer demand;
+  private boolean initialConsumeCalled;
+  private Runnable initialConsume;
 
   JettyAsyncResponseListener(HttpRequest httpRequest, HttpClient.BodyConsumer<T> bodyConsumer) {
     this.httpRequest = httpRequest;
     this.bodyConsumer = bodyConsumer;
     asyncResponse = new CompletableFuture<>();
     asyncBodyDone = new CompletableFuture<>();
-    consume = false;
   }
 
   @Override
-  public synchronized void consume() {
-    consume = true;
-    this.notifyAll();
+  public void consume() {
+    synchronized (this) {
+      if (!this.initialConsumeCalled) {
+        this.initialConsumeCalled = true;
+        if (this.initialConsume != null) {
+          this.initialConsume.run();
+          this.initialConsume = null;
+        }
+      }
+    }
+    demand.accept(1);
   }
 
   @Override
@@ -74,21 +84,20 @@ public CompletableFuture<HttpResponse<HttpClient.AsyncBody>> listen(Request requ
   }
 
   @Override
-  public void onContent(Response response, ByteBuffer content) {
-    try {
-      synchronized (this) {
-        while (!consume && !asyncBodyDone.isCancelled()) {
-          this.wait();
-        }
-      }
-      if (!asyncBodyDone.isCancelled()) {
-        bodyConsumer.consume(process(response, content), this);
+  public void onContent(Response response, LongConsumer demand, ByteBuffer content, Callback callback) {
+    synchronized (this) {
+      if (!initialConsumeCalled) {
+        // defer until consume is called
+        this.initialConsume = () -> onContent(response, demand, content, callback);
+        return;
       }
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-      throw KubernetesClientException.launderThrowable(e);
+      this.demand = demand;
+    }
+    try {
+      bodyConsumer.consume(process(response, content), this);
+      callback.succeeded();
     } catch (Exception e) {
-      throw KubernetesClientException.launderThrowable(e);
+      callback.failed(e);
     }
   }
 

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

@@ -72,16 +72,21 @@ public void consume() {
       // consume should not block from a callers perspective
       try {
         httpClient.dispatcher().executorService().execute(() -> {
-          try {
-            if (!source.exhausted() && !done.isDone()) {
-              T value = process(source);
-              consumer.consume(value, this);
-            } else {
-              done.complete(null);
+          // we must serialize multiple consumes as source is not thread-safe
+          // it would be better to use SerialExecutor, but that would need to move modules, as to
+          // potentially not hold multiple executor threads
+          synchronized (source) {
+            try {
+              if (!source.exhausted() && !done.isDone()) {
+                T value = process(source);
+                consumer.consume(value, this);
+              } else {
+                done.complete(null);
+              }
+            } catch (Exception e) {
+              Utils.closeQuietly(source);
+              done.completeExceptionally(e);
             }
-          } catch (Exception e) {
-            Utils.closeQuietly(source);
-            done.completeExceptionally(e);
           }
         });
       } catch (Exception e) {

httpclient-tests/src/test/java/io/fabric8/kubernetes/client/http/OkHttpClientTest.java

@@ -31,12 +31,15 @@
 import java.io.InputStream;
 import java.io.Reader;
 import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -53,8 +56,8 @@ protected HttpClient.Factory getHttpClientFactory() {
     return new OkHttpClientFactory();
   }
 
-  private KubernetesMockServer server;
-  private KubernetesClient client;
+  KubernetesMockServer server;
+  KubernetesClient client;
 
   @BeforeEach
   void setUp() {
@@ -148,15 +151,17 @@ public void onMessage(WebSocket webSocket, String text) {
 
   @Test
   void testAsyncBody() throws Exception {
-    server.expect().withPath("/async").andReturn(200, "hello world").always();
+    int byteCount = 20000;
+    server.expect().withPath("/async").andReturn(200, new String(new byte[byteCount], StandardCharsets.UTF_8)).always();
 
-    CompletableFuture<Boolean> consumed = new CompletableFuture<>();
+    CompletableFuture<Integer> consumed = new CompletableFuture<>();
+    AtomicInteger total = new AtomicInteger();
 
     CompletableFuture<HttpResponse<AsyncBody>> responseFuture = client.getHttpClient().consumeBytes(
         client.getHttpClient().newHttpRequestBuilder().uri(URI.create(client.getConfiguration().getMasterUrl() + "async"))
             .build(),
         (value, asyncBody) -> {
-          consumed.complete(true);
+          value.stream().map(ByteBuffer::remaining).forEach(total::addAndGet);
           asyncBody.consume();
         });
 
@@ -169,15 +174,14 @@ void testAsyncBody() throws Exception {
         r.body().done().whenComplete((v, ex) -> {
           if (ex != null) {
             consumed.completeExceptionally(ex);
-          }
-          if (v != null) {
-            consumed.complete(false);
+          } else {
+            consumed.complete(total.get());
           }
         });
       }
     });
 
-    assertTrue(consumed.get(5, TimeUnit.SECONDS));
+    assertEquals(byteCount, consumed.get(5, TimeUnit.SECONDS));
   }
 
   @Test
@@ -219,14 +223,15 @@ void testConsumeLines() throws Exception {
   @ParameterizedTest(name = "{index}: {0}")
   @ValueSource(classes = { String.class, byte[].class, Reader.class, InputStream.class })
   void testSupportedTypes(Class<?> type) throws Exception {
-    server.expect().withPath("/type").andReturn(200, "hello world").always();
+    String value = new String(new byte[16384]);
+    server.expect().withPath("/type").andReturn(200, value).always();
     final HttpResponse<?> result = client.getHttpClient()
         .sendAsync(client.getHttpClient().newHttpRequestBuilder()
             .uri(URI.create(client.getConfiguration().getMasterUrl() + "type")).build(), type)
         .get(10, TimeUnit.SECONDS);
     assertThat(result)
         .satisfies(r -> assertThat(r.body()).isInstanceOf(type))
-        .satisfies(r -> assertThat(r.bodyString()).isEqualTo("hello world"));
+        .satisfies(r -> assertThat(r.bodyString()).isEqualTo(value));
   }
 
 }