replaced wait/notify with countdownlatch and initiated websocket closure from the client side.

sagarbhavsar4328 · sagarbhavsar4328 · commit df2dbb40371e · 2020-04-27T13:36:36.000+05:30
diff --git a/util/src/main/java/io/kubernetes/client/Exec.java b/util/src/main/java/io/kubernetes/client/Exec.java
@@ -37,6 +37,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
@@ -354,6 +355,7 @@ public static class ExecProcess extends Process {
     private int statusCode = -1;
     private boolean isAlive = true;
     private final Map<Integer, InputStream> input = new HashMap<>();
+    private final CountDownLatch latch = new CountDownLatch(1);
 
     public ExecProcess(final ApiClient apiClient) throws IOException {
       this.streamHandler =
@@ -367,10 +369,13 @@ protected void handleMessage(int stream, InputStream inStream) throws IOExceptio
                   synchronized (ExecProcess.this) {
                     statusCode = exitCode;
                     isAlive = false;
-                    ExecProcess.this.notifyAll();
                   }
                 }
                 inStream.close();
+                // Stream ID of `3` delivers the status of exec connection from kubelet,
+                // closing the connection upon 0 exit-code.
+                this.close();
+                ExecProcess.this.latch.countDown();
               } else super.handleMessage(stream, inStream);
             }
 
@@ -386,7 +391,7 @@ public void failure(Throwable ex) {
                 // code.
                 statusCode = -1975219;
                 isAlive = false;
-                ExecProcess.this.notifyAll();
+                ExecProcess.this.latch.countDown();
               }
             }
 
@@ -396,7 +401,7 @@ public void close() {
               synchronized (ExecProcess.this) {
                 if (isAlive) {
                   isAlive = false;
-                  ExecProcess.this.notifyAll();
+                  ExecProcess.this.latch.countDown();
                 }
               }
 
@@ -442,18 +447,14 @@ private synchronized InputStream getInputStream(int stream) {
 
     @Override
     public int waitFor() throws InterruptedException {
-      synchronized (this) {
-        this.wait();
-        return statusCode;
-      }
+      this.latch.await();
+      return statusCode;
     }
 
     @Override
     public boolean waitFor(long timeout, TimeUnit unit) throws InterruptedException {
-      synchronized (this) {
-        this.wait(TimeUnit.MILLISECONDS.convert(timeout, unit));
-        return !isAlive();
-      }
+      this.latch.await(timeout, unit);
+      return !isAlive();
     }
 
     @Override
diff --git a/util/src/test/java/io/kubernetes/client/ExecTest.java b/util/src/test/java/io/kubernetes/client/ExecTest.java
@@ -109,14 +109,14 @@ public void testExecProcess() throws IOException, InterruptedException {
 
     process.getHandler().bytesMessage(makeStream(1, msgData.getBytes(StandardCharsets.UTF_8)));
     process.getHandler().bytesMessage(makeStream(2, errData.getBytes(StandardCharsets.UTF_8)));
-    process.getHandler().bytesMessage(makeStream(3, OUTPUT_EXIT0.getBytes(StandardCharsets.UTF_8)));
 
     final ByteArrayOutputStream stdout = new ByteArrayOutputStream();
     final ByteArrayOutputStream stderr = new ByteArrayOutputStream();
 
     Thread t1 = asyncCopy(process.getInputStream(), stdout);
     Thread t2 = asyncCopy(process.getErrorStream(), stderr);
 
+    process.getHandler().bytesMessage(makeStream(3, OUTPUT_EXIT0.getBytes(StandardCharsets.UTF_8)));
     // TODO: Fix this asap!
     Thread.sleep(1000);
 
