Additional race conditions and resolve code review comments

Author: rjeberhard

examples/src/main/java/io/kubernetes/client/examples/ExecExample.java

@@ -1,5 +1,5 @@
 /*
-Copyright 2017 The Kubernetes Authors.
+Copyright 2017, 2018 The Kubernetes Authors.
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
@@ -19,6 +19,9 @@
 import io.kubernetes.client.Exec;
 import io.kubernetes.client.util.Config;
 import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * A simple example of how to use the Java API
@@ -30,47 +33,68 @@
  */
 public class ExecExample {
   public static void main(String[] args) throws IOException, ApiException, InterruptedException {
+    String podName = "nginx-4217019353-k5sn9";
+    String namespace = "default";
+    List<String> commands = new ArrayList<>();
+
+    int len = args.length;
+    if (len >= 1) podName = args[0];
+    if (len >= 2) namespace = args[1];
+    for (int i = 2; i < len; i++) {
+      commands.add(args[i]);
+    }
+
     ApiClient client = Config.defaultClient();
     Configuration.setDefaultApiClient(client);
 
     Exec exec = new Exec();
-    boolean tty = System.console() != null;
-    //        final Process proc = exec.exec("default", "nginx-2371676037-czqx3", new String[]
-    // {"sh", "-c", "echo foo"}, true, tty);
+    boolean tty = true; // Optional: System.console() != null;
+    // final Process proc = exec.exec("default", "nginx-4217019353-k5sn9", new String[]
+    //   {"sh", "-c", "echo foo"}, true, tty);
     final Process proc =
-        exec.exec("default", "nginx-4217019353-k5sn9", new String[] {"sh"}, true, tty);
+        exec.exec(
+            namespace,
+            podName,
+            commands.isEmpty()
+                ? new String[] {"sh"}
+                : commands.toArray(new String[commands.size()]),
+            true,
+            tty);
 
-    new Thread(
+    Thread in =
+        new Thread(
             new Runnable() {
               public void run() {
                 try {
                   ByteStreams.copy(System.in, proc.getOutputStream());
+                } catch (InterruptedIOException ie) {
+                  // no-op
                 } catch (IOException ex) {
                   ex.printStackTrace();
                 }
               }
-            })
-        .start();
+            });
+    in.start();
 
-    new Thread(
+    Thread out =
+        new Thread(
             new Runnable() {
               public void run() {
                 try {
                   ByteStreams.copy(proc.getInputStream(), System.out);
+                } catch (InterruptedIOException ie) {
+                  // no-op
                 } catch (IOException ex) {
                   ex.printStackTrace();
                 }
               }
-            })
-        .start();
+            });
+    out.start();
 
     proc.waitFor();
-    try {
-      // Wait for buffers to flush.
-      Thread.sleep(2000);
-    } catch (InterruptedException ex) {
-      ex.printStackTrace();
-    }
+
+    // wait for any last output; no need to wait for input thread
+    out.join();
 
     proc.destroy();
 

util/src/main/java/io/kubernetes/client/Exec.java

@@ -1,5 +1,5 @@
 /*
-Copyright 2017 The Kubernetes Authors.
+Copyright 2017, 2018 The Kubernetes Authors.
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
@@ -18,10 +18,16 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class Exec {
+  private static final Logger log = LoggerFactory.getLogger(Exec.class);
+
   private ApiClient apiClient;
 
   /** Simple Exec API constructor, uses default configuration */
@@ -178,10 +184,34 @@ public Process exec(
     return exec;
   }
 
+  static int parseExitCode(InputStream inputStream) {
+    int exitCode = 0;
+    try {
+      int available = inputStream.available();
+      if (available > 0) {
+        byte[] b = new byte[available];
+        inputStream.read(b);
+        String result = new String(b, "UTF-8");
+        int idx = result.lastIndexOf(':');
+        if (idx > 0) {
+          try {
+            exitCode = Integer.parseInt(result.substring(idx + 1).trim());
+          } catch (NumberFormatException nfe) {
+            log.error("Error parsing exit code from status channel response", nfe);
+          }
+        }
+      }
+    } catch (IOException io) {
+      log.error("Error parsing exit code from status channel response", io);
+    }
+
+    return exitCode;
+  }
+
   private static class ExecProcess extends Process {
     private final WebSocketStreamHandler streamHandler;
     private volatile int statusCode;
-    private volatile boolean isDestroyed = false;
+    private final Map<Integer, InputStream> input = new HashMap<>();
 
     public ExecProcess() throws IOException {
       this.statusCode = -1;
@@ -190,25 +220,7 @@ public ExecProcess() throws IOException {
             @Override
             public void close() {
               if (statusCode == -1) {
-                InputStream inputStream = getInputStream(3);
-                int exitCode = 0;
-                try {
-                  int available = inputStream.available();
-                  if (available > 0) {
-                    byte[] b = new byte[available];
-                    inputStream.read(b);
-                    String result = new String(b, "UTF-8");
-                    int idx = result.lastIndexOf(':');
-                    if (idx > 0) {
-                      try {
-                        exitCode = Integer.parseInt(result.substring(idx + 1).trim());
-                      } catch (NumberFormatException nfe) {
-                        // no-op
-                      }
-                    }
-                  }
-                } catch (IOException e) {
-                }
+                int exitCode = parseExitCode(ExecProcess.this.getInputStream(3));
 
                 // notify of process completion
                 synchronized (ExecProcess.this) {
@@ -217,7 +229,7 @@ public void close() {
                 }
               }
 
-              if (isDestroyed) super.close();
+              super.close();
             }
           };
     }
@@ -233,12 +245,19 @@ public OutputStream getOutputStream() {
 
     @Override
     public InputStream getInputStream() {
-      return streamHandler.getInputStream(1);
+      return getInputStream(1);
     }
 
     @Override
     public InputStream getErrorStream() {
-      return streamHandler.getInputStream(2);
+      return getInputStream(2);
+    }
+
+    private synchronized InputStream getInputStream(int stream) {
+      if (!input.containsKey(stream)) {
+        input.put(stream, streamHandler.getInputStream(stream));
+      }
+      return input.get(stream);
     }
 
     @Override
@@ -265,8 +284,14 @@ public int exitValue() {
 
     @Override
     public void destroy() {
-      isDestroyed = true;
       streamHandler.close();
+      for (InputStream in : input.values()) {
+        try {
+          in.close();
+        } catch (IOException ex) {
+          log.error("Error on close", ex);
+        }
+      }
     }
   }
 }