fix fabric8io#4910 fabric8io#4923: addressing vertx and general race …

…condition with exec

CHANGELOG.md

@@ -23,6 +23,7 @@
 * Fix #4931: using coarse grain locking for all mock server operations
 * Fix #4947: typo in HttpClient.Factory scoring system logic
 * Fix #4928: allows non-okhttp clients to handle invalid status
+* Fix #4910 / #4923: addressing inconsistent behavior with pod exec operations
 
 #### Improvements
 * Fix #4675: adding a fully client side timeout for informer watches

httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/VertxWebSocket.java

@@ -44,7 +44,14 @@ void init() {
       ws.pause();
       listener.onMessage(this, msg);
     });
-    ws.closeHandler(v -> listener.onClose(this, ws.closeStatusCode(), ws.closeReason()));
+    // if the server sends a ping, we're in trouble with our fetch strategy as there is
+    // no ping handler to increase the demand - this should not be an immediate issue as
+    // the api server does not seem to be sending pings
+
+    // if for whatever reason we send a ping, pong counts against the demand, so we need more
+    ws.pongHandler(b -> ws.fetch(1));
+    // use end, not close, because close is processed immediately vs. end is in frame order
+    ws.endHandler(v -> listener.onClose(this, ws.closeStatusCode(), ws.closeReason()));
     ws.exceptionHandler(err -> listener.onError(this, err));
     listener.onOpen(this);
   }

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

@@ -66,6 +66,7 @@ void consume(List<ByteBuffer> value) {
         request.run();
         return;
       }
+      assert !complete || failed == null;
       buffers.addAll(value);
       buffers.notifyAll();
     }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/core/v1/PodOperationsImpl.java

@@ -498,15 +498,16 @@ private InputStream read(String... command) {
   private Future<?> readTo(OutputStream out, String... cmd) {
     ExecWatch w = writingOutput(out).exec(cmd);
     CompletableFuture<Integer> result = w.exitCode();
-    result.whenComplete((i, t) -> {
+    // ensure the whenComplete runs prior to downstream actions
+    return result.whenComplete((i, t) -> {
       try {
         out.close();
       } catch (Exception e) {
         result.obtrudeException(e);
+      } finally {
+        w.close();
       }
-      w.close();
     });
-    return result;
   }
 
   private void copyDir(String source, File target) throws Exception {

kubernetes-itests/src/test/java/io/fabric8/kubernetes/PodIT.java

@@ -348,7 +348,7 @@ void copyDir() throws IOException {
     PodResource podResource = client.pods().withName("pod-standard");
     podResource.dir("/etc").withReadyWaitTimeout(POD_READY_WAIT_IN_MILLIS).copy(tmpDir);
 
-    Path msg = tmpDir.resolve("/etc/hosts");
+    Path msg = tmpDir.resolve("etc/hosts");
     assertTrue(Files.exists(msg));
   }