fix fabric8io#4885: addressing a potential hang with jdk streams

CHANGELOG.md

@@ -14,6 +14,7 @@
 * Fix #4818: [java-generator] Escape `*/` in generated JavaDocs
 * Fix #4823: (java-generator) handle special characters in field names
 * Fix #4723: [java-generator] Fix a race in the use of JavaParser hitting large CRDs
+* Fix #4885: addresses a potential hang in the jdk client with exec stream reading
 
 #### Improvements
 * Fix #3805: DeletionTimestamp and Finalizer support in Mock server.

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

@@ -86,6 +86,9 @@ public void onSubscribe(Subscription subscription) {
 
         @Override
         public void onNext(List<ByteBuffer> item) {
+          // there doesn't seem to be a guarantee that the buffer won't be modified by the caller
+          // after passing it in, so we'll create a copy
+          //bodySubscriber.onNext(item.stream().map(BufferUtil::copy).collect(Collectors.toList()));
           bodySubscriber.onNext(item);
         }
 

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

@@ -90,7 +90,7 @@ private ByteBuffer current() throws IOException {
 
         currentBuffer = buffers.poll();
 
-        if (currentBuffer == null) {
+        if (currentBuffer == null && !complete) {
           try {
             buffers.wait();
           } catch (InterruptedException e) {

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

@@ -91,6 +91,9 @@ public LogWatchCallback callAndWait(HttpClient client, URL url) {
         for (ByteBuffer byteBuffer : buffers) {
           try {
             outChannel.write(byteBuffer);
+            if (!closed.get()) {
+              a.consume();
+            }
           } catch (IOException e1) {
             throw KubernetesClientException.launderThrowable(e1);
           }
@@ -99,10 +102,6 @@ public LogWatchCallback callAndWait(HttpClient client, URL url) {
         if (t != null) {
           a.cancel();
           onFailure(t);
-        } else if (!closed.get()) {
-          a.consume();
-        } else {
-          a.cancel();
         }
       })).whenComplete((a, e) -> {
         if (e != null) {

kubernetes-client/src/test/java/io/fabric8/kubernetes/client/dsl/internal/ExecWatchInputStreamTest.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -93,4 +94,15 @@ void testConsume() throws IOException {
     readFuture.join();
   }
 
+  @Test
+  void testCompleteInlineWithRequestMore() throws IOException {
+    AtomicReference<ExecWatchInputStream> is = new AtomicReference<>();
+    ExecWatchInputStream execWatchInputStream = new ExecWatchInputStream(() -> is.get().onExit(0, null));
+    is.set(execWatchInputStream);
+    // the first consume is implicit
+    execWatchInputStream.consume(Collections.singletonList(ByteBuffer.allocate(1)));
+    is.get().read();
+    Awaitility.await().atMost(1, TimeUnit.SECONDS).until(() -> execWatchInputStream.read() == -1);
+  }
+
 }