Fix resource lifecycle issues in KubectlPortForward

Without this change only one instance of PortForward is ever created and so when its resources are closed they cannot be re-used. It is better to create a new instance for each local socket, and then second and subsequent client connections will succeed.

Author: Dave Syer

extended/src/main/java/io/kubernetes/client/extended/kubectl/KubectlPortForward.java

@@ -24,6 +24,7 @@
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Consumer;
@@ -78,47 +79,46 @@ public void shutdown() {
  private void executeInternal()
  throws ApiException, KubectlException, IOException, InterruptedException {
  PortForward pf = new PortForward(apiClient);
- PortForward.PortForwardResult result = pf.forward(namespace, name, targetPorts);
- if (result == null) {
- throw new KubectlException("PortForward failed!");
- }
  // TODO: Convert this to NIO to reduce the number of threads?
  List<Thread> threads = new ArrayList<>();
  for (int i = 0; i < localPorts.size(); i++) {
  int targetPort = targetPorts.get(i);
  threads.add(
- portForward(
- new ServerSocket(localPorts.get(i)),
- result.getInputStream(targetPort),
- result.getOutboundStream(targetPort)));
+ portForward(pf,
+ new ServerSocket(localPorts.get(i)), targetPort));
  }
  for (Thread t : threads) {
  t.join();
  }
  }
 
- private Thread portForward(ServerSocket server, InputStream in, OutputStream out) {
- Thread t =
- new Thread(
- new Runnable() {
- @Override
- public void run() {
- while (running) {
- try {
- Socket sock = server.accept();
- Thread t1 = copyAsync(sock.getInputStream(), out, onUnhandledError);
- Thread t2 = copyAsync(in, sock.getOutputStream(), onUnhandledError);
-
- t1.join();
- t2.join();
- } catch (InterruptedException | IOException ex) {
- Optional.ofNullable(onUnhandledError)
- .orElse(Throwable::printStackTrace)
- .accept(ex);
- }
+ private Thread portForward(PortForward pf, ServerSocket server, int targetPort) {
+ Thread t = new Thread(
+ new Runnable() {
+ @Override
+ public void run() {
+ while (running) {
+ try (Socket sock = server.accept()) {
+ PortForward.PortForwardResult result = pf.forward(namespace, name, Arrays.asList(targetPort));
+ if (result == null) {
+ throw new KubectlException("PortForward failed!");
  }
+ InputStream in = result.getInputStream(targetPort);
+ OutputStream out = result.getOutboundStream(targetPort);
+ Thread t1 = copyAsync(sock.getInputStream(), out, onUnhandledError);
+ Thread t2 = copyAsync(in, sock.getOutputStream(), onUnhandledError);
+
+ t1.join();
+ in.close();
+ t2.join();
+ } catch (Exception ex) {
+ Optional.ofNullable(onUnhandledError)
+ .orElse(Throwable::printStackTrace)
+ .accept(ex);
  }
- });
+ }
+ }
+ });
  t.start();
  return t;
  }