fix #4885: addressing a potential hang with jdk streams

CHANGELOG.md

@@ -13,6 +13,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.

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

@@ -86,7 +86,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/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;
@@ -85,4 +86,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);
+  }
+
 }