Fix jetty completion handling to forward failures

CHANGELOG.md

@@ -6,6 +6,7 @@
 * Fix #6038: Support for Gradle configuration cache
 * Fix #6110: VolumeSource (and other file mode fields) in Octal are correctly interpreted
 * Fix #6215: Suppressing rejected execution exception for port forwarder
+* Fix #6197: JettyHttp client error handling improvements. 
 
 #### Improvements
 * Fix #6008: removing the optional dependency on bouncy castle

httpclient-jdk/src/test/java/io/fabric8/kubernetes/client/jdkhttp/JdkHttpClientPostTest.java

@@ -18,6 +18,8 @@
 import io.fabric8.kubernetes.client.http.AbstractHttpPostTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
 
+import java.io.IOException;
+
 @SuppressWarnings("java:S2187")
 public class JdkHttpClientPostTest extends AbstractHttpPostTest {
   @Override
@@ -31,4 +33,9 @@ public void expectContinue() {
     // Apparently the JDK sets the Expect header to 100-Continue which is not according to spec.
     // We should consider overriding the header manually instead.
   }
+
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    return IOException.class;
+  }
 }

httpclient-jdk/src/test/java/io/fabric8/kubernetes/client/jdkhttp/JdkHttpClientPutTest.java

@@ -18,10 +18,17 @@
 import io.fabric8.kubernetes.client.http.AbstractHttpPutTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
 
+import java.io.IOException;
+
 @SuppressWarnings("java:S2187")
 public class JdkHttpClientPutTest extends AbstractHttpPutTest {
   @Override
   protected HttpClient.Factory getHttpClientFactory() {
     return new JdkHttpClientFactory();
   }
+
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    return IOException.class;
+  }
 }

httpclient-jetty/src/main/java/io/fabric8/kubernetes/client/jetty/JettyAsyncResponseListener.java

@@ -72,7 +72,16 @@ public void onHeaders(Response response) {
 
   @Override
   public void onComplete(Result result) {
-    asyncBodyDone.complete(null);
+    if (result.isSucceeded()) {
+      asyncBodyDone.complete(null);
+    } else {
+      asyncBodyDone.completeExceptionally(result.getRequestFailure());
+    }
+  }
+
+  @Override
+  public void onFailure(Response response, Throwable failure) {
+    asyncResponse.completeExceptionally(failure);
   }
 
   public CompletableFuture<HttpResponse<AsyncBody>> listen(Request request) {

httpclient-jetty/src/test/java/io/fabric8/kubernetes/client/jetty/JettyHttpPostTest.java

@@ -18,10 +18,17 @@
 import io.fabric8.kubernetes.client.http.AbstractHttpPostTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
 
+import java.net.ConnectException;
+
 @SuppressWarnings("java:S2187")
 public class JettyHttpPostTest extends AbstractHttpPostTest {
   @Override
   protected HttpClient.Factory getHttpClientFactory() {
     return new JettyHttpClientFactory();
   }
+
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    return ConnectException.class;
+  }
 }

httpclient-jetty/src/test/java/io/fabric8/kubernetes/client/jetty/JettyHttpPutTest.java

@@ -18,10 +18,17 @@
 import io.fabric8.kubernetes.client.http.AbstractHttpPutTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
 
+import java.net.ConnectException;
+
 @SuppressWarnings("java:S2187")
 public class JettyHttpPutTest extends AbstractHttpPutTest {
   @Override
   protected HttpClient.Factory getHttpClientFactory() {
     return new JettyHttpClientFactory();
   }
+
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    return ConnectException.class;
+  }
 }

httpclient-okhttp/src/test/java/io/fabric8/kubernetes/client/okhttp/OkHttpPostTest.java

@@ -17,11 +17,24 @@
 
 import io.fabric8.kubernetes.client.http.AbstractHttpPostTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
+import org.junit.jupiter.api.condition.OS;
+
+import java.io.InterruptedIOException;
+import java.net.ConnectException;
 
 @SuppressWarnings("java:S2187")
 public class OkHttpPostTest extends AbstractHttpPostTest {
   @Override
   protected HttpClient.Factory getHttpClientFactory() {
     return new OkHttpClientFactory();
   }
+
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    if (OS.WINDOWS.equals(OS.current())) {
+      return InterruptedIOException.class;
+    } else {
+      return ConnectException.class;
+    }
+  }
 }

httpclient-okhttp/src/test/java/io/fabric8/kubernetes/client/okhttp/OkHttpPutTest.java

@@ -17,11 +17,24 @@
 
 import io.fabric8.kubernetes.client.http.AbstractHttpPutTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
+import org.junit.jupiter.api.condition.OS;
+
+import java.io.InterruptedIOException;
+import java.net.ConnectException;
 
 @SuppressWarnings("java:S2187")
 public class OkHttpPutTest extends AbstractHttpPutTest {
   @Override
   protected HttpClient.Factory getHttpClientFactory() {
     return new OkHttpClientFactory();
   }
+
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    if (OS.WINDOWS.equals(OS.current())) {
+      return InterruptedIOException.class;
+    } else {
+      return ConnectException.class;
+    }
+  }
 }

httpclient-vertx/src/test/java/io/fabric8/kubernetes/client/vertx/VertxHttpClientPostTest.java

@@ -18,11 +18,18 @@
 import io.fabric8.kubernetes.client.http.AbstractHttpPostTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
 
+import java.net.ConnectException;
+
 @SuppressWarnings("java:S2187")
 public class VertxHttpClientPostTest extends AbstractHttpPostTest {
   @Override
   protected HttpClient.Factory getHttpClientFactory() {
     return new VertxHttpClientFactory();
   }
 
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    return ConnectException.class;
+  }
+
 }

httpclient-vertx/src/test/java/io/fabric8/kubernetes/client/vertx/VertxHttpClientPutTest.java

@@ -18,10 +18,17 @@
 import io.fabric8.kubernetes.client.http.AbstractHttpPutTest;
 import io.fabric8.kubernetes.client.http.HttpClient;
 
+import java.net.ConnectException;
+
 @SuppressWarnings("java:S2187")
 public class VertxHttpClientPutTest extends AbstractHttpPutTest {
   @Override
   protected HttpClient.Factory getHttpClientFactory() {
     return new VertxHttpClientFactory();
   }
+
+  @Override
+  protected Class<? extends Exception> getConnectionFailedExceptionType() {
+    return ConnectException.class;
+  }
 }

junit/mockwebserver/src/main/java/io/fabric8/mockwebserver/internal/MockDispatcher.java

@@ -21,6 +21,7 @@
 import okhttp3.mockwebserver.Dispatcher;
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.RecordedRequest;
+import okhttp3.mockwebserver.SocketPolicy;
 
 import java.util.Collection;
 import java.util.Map;
@@ -36,6 +37,11 @@ public MockDispatcher(Map<ServerRequest, Queue<ServerResponse>> responses) {
     this.responses = responses;
   }
 
+  @Override
+  public MockResponse peek() {
+    return new MockResponse().setSocketPolicy(SocketPolicy.EXPECT_CONTINUE);
+  }
+
   @Override
   public MockResponse dispatch(RecordedRequest request) {
     for (WebSocketSession webSocketSession : webSocketSessions) {

kubernetes-client-api/src/test/java/io/fabric8/kubernetes/client/http/AbstractHttpPostTest.java

@@ -23,8 +23,15 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -46,6 +53,8 @@ static void afterAll() {
 
   protected abstract HttpClient.Factory getHttpClientFactory();
 
+  protected abstract Class<? extends Exception> getConnectionFailedExceptionType();
+
   @Test
   @DisplayName("String body, should send a POST request with body")
   public void postStringBody() throws Exception {
@@ -128,20 +137,53 @@ public void postFormDataBody() throws Exception {
 
   @Test
   public void expectContinue() throws Exception {
+    server.expect().post().withPath("/post-expect-continue").andReturn(200, "").always();
+
     // When
     try (HttpClient client = getHttpClientFactory().newBuilder().build()) {
-      client
+      final CompletableFuture<HttpResponse<String>> response = client
           .sendAsync(client.newHttpRequestBuilder()
               .post(Collections.emptyMap())
               .uri(server.url("/post-expect-continue"))
               .expectContinue()
-              .build(), String.class)
-          .get(10L, TimeUnit.SECONDS);
+              .build(), String.class);
+
+      // Then
+      assertThat(response).succeedsWithin(10, TimeUnit.SECONDS);
+      assertThat(server.getLastRequest())
+          .returns("POST", RecordedRequest::getMethod)
+          .extracting(rr -> rr.getHeader("Expect")).asString()
+          .isEqualTo("100-continue");
     }
-    // Then
-    assertThat(server.getLastRequest())
-        .returns("POST", RecordedRequest::getMethod)
-        .extracting(rr -> rr.getHeader("Expect")).asString()
-        .isEqualTo("100-continue");
   }
+
+  @Test
+  public void expectFailure() throws IOException, URISyntaxException {
+    try (final ServerSocket serverSocket = new ServerSocket(0)) {
+
+      try (HttpClient client = getHttpClientFactory().newBuilder().build()) {
+        final URI uri = uriForPath(serverSocket, "/post-failing");
+        serverSocket.close();
+
+        // When
+        final CompletableFuture<HttpResponse<String>> response = client
+            .sendAsync(client.newHttpRequestBuilder()
+                .post("text/plain", new ByteArrayInputStream("A string body".getBytes(StandardCharsets.UTF_8)), -1)
+                .uri(uri)
+                .timeout(250, TimeUnit.MILLISECONDS)
+                .build(), String.class);
+
+        // Then
+        assertThat(response).failsWithin(30, TimeUnit.SECONDS)
+            .withThrowableOfType(ExecutionException.class)
+            .withCauseInstanceOf(getConnectionFailedExceptionType());
+      }
+    }
+  }
+
+  private static URI uriForPath(ServerSocket socket, String path) throws URISyntaxException {
+    final InetAddress httpServerAddress = socket.getInetAddress();
+    return new URI(String.format("http://%s:%s%s", httpServerAddress.getHostName(), socket.getLocalPort(), path));
+  }
+
 }

kubernetes-client-api/src/test/java/io/fabric8/kubernetes/client/http/AbstractHttpPutTest.java

@@ -23,7 +23,14 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -44,6 +51,8 @@ static void afterAll() {
 
   protected abstract HttpClient.Factory getHttpClientFactory();
 
+  protected abstract Class<? extends Exception> getConnectionFailedExceptionType();
+
   @Test
   @DisplayName("String body, should send a PUT request with body")
   public void putStringBody() throws Exception {
@@ -83,4 +92,33 @@ public void putInputStreamBody() throws Exception {
         .extracting(rr -> rr.getHeader("Content-Type")).asString()
         .startsWith("text/plain");
   }
+
+  @Test
+  public void expectFailure() throws IOException, URISyntaxException {
+    try (final ServerSocket serverSocket = new ServerSocket(0)) {
+
+      try (HttpClient client = getHttpClientFactory().newBuilder().build()) {
+        final URI uri = uriForPath(serverSocket, "/put-failing");
+        serverSocket.close();
+
+        // When
+        final CompletableFuture<HttpResponse<String>> response = client
+            .sendAsync(client.newHttpRequestBuilder()
+                .put("text/plain", new ByteArrayInputStream("A string body".getBytes(StandardCharsets.UTF_8)), -1)
+                .uri(uri)
+                .timeout(250, TimeUnit.MILLISECONDS)
+                .build(), String.class);
+
+        // Then
+        assertThat(response).failsWithin(30, TimeUnit.SECONDS)
+            .withThrowableOfType(ExecutionException.class)
+            .withCauseInstanceOf(getConnectionFailedExceptionType());
+      }
+    }
+  }
+
+  private static URI uriForPath(ServerSocket socket, String path) throws URISyntaxException {
+    final InetAddress httpServerAddress = socket.getInetAddress();
+    return new URI(String.format("http://%s:%s%s", httpServerAddress.getHostName(), socket.getLocalPort(), path));
+  }
 }