Introduce afterConnectionFailure to complement after & afterFailure.

Fixes fabric8io#6143

kubernetes-client-api/src/main/java/io/fabric8/kubernetes/client/http/Interceptor.java

@@ -84,4 +84,15 @@ default CompletableFuture<Boolean> afterFailure(HttpRequest.Builder builder, Htt
     return afterFailure((BasicBuilder) builder, response, tags);
   }
 
+  /**
+   * Called after a connection attempt fails.
+   * <p>
+   * This method will be invoked on each failed connection attempt so can be invoked multiple times for the same request.ID.
+   *
+   * @param request the HTTP request.
+   * @param failure the Java exception that caused the failure.
+   */
+  default void afterConnectionFailure(HttpRequest request, Throwable failure) {
+  }
+
 }

kubernetes-client-api/src/main/java/io/fabric8/kubernetes/client/http/StandardHttpClient.java

@@ -108,6 +108,13 @@ private CompletableFuture<HttpResponse<AsyncBody>> consumeBytesOnce(StandardHttp
     final Consumer<List<ByteBuffer>> effectiveConsumer = consumer;
 
     CompletableFuture<HttpResponse<AsyncBody>> cf = consumeBytesDirect(effectiveRequest, effectiveConsumer);
+    cf.exceptionally(throwable -> {
+      builder.interceptors.forEach((s, interceptor) -> interceptor.afterConnectionFailure(effectiveRequest, throwable));
+      if (throwable instanceof CompletionException) {
+        throw (CompletionException) throwable;
+      }
+      throw new CompletionException(throwable);
+    });
     cf.thenAccept(
         response -> builder.getInterceptors().values().forEach(i -> i.after(effectiveRequest, response, effectiveConsumer)));
 

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

@@ -32,6 +32,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -173,24 +174,62 @@ public CompletableFuture<Boolean> afterFailure(BasicBuilder builder, HttpRespons
   }
 
   @Test
-  @DisplayName("afterFailure (HTTP), invoked when remote server offline")
-  public void afterHttpFailureRemoteOffline() {
+  @DisplayName("afterConnectionFailure, invoked when remote server offline")
+  public void afterConnectionFailureRemoteOffline() {
     // Given
     server.shutdown();
+    final CountDownLatch connectionFailureCallbackInvoked = new CountDownLatch(1);
     final HttpClient.Builder builder = getHttpClientFactory().newBuilder()
         .connectTimeout(1, TimeUnit.SECONDS)
         .addOrReplaceInterceptor("test", new Interceptor() {
           @Override
-          public CompletableFuture<Boolean> afterFailure(BasicBuilder builder, HttpResponse<?> response, RequestTags tags) {
-            return CompletableFuture.completedFuture(false);
+          public void afterConnectionFailure(HttpRequest request, Throwable failure) {
+            connectionFailureCallbackInvoked.countDown();
+          }
+        });
+    // When
+    try (HttpClient client = builder.build()) {
+      final CompletableFuture<HttpResponse<String>> response = client.sendAsync(client.newHttpRequestBuilder()
+          .timeout(1, TimeUnit.SECONDS)
+          .uri(server.url("/not-found")).build(), String.class);
+
+      // Then
+      assertThat(response).failsWithin(Duration.of(30, ChronoUnit.SECONDS));
+      assertThat(connectionFailureCallbackInvoked).extracting(CountDownLatch::getCount).isEqualTo(0L);
+    }
+  }
+
+  @Test
+  @DisplayName("afterConnectionFailure, request is retried when remote server offline")
+  public void afterConnectionFailureRetry() {
+    // Given
+    final int originalPort = server.getPort();
+    server.shutdown();
+    final CountDownLatch afterInvoked = new CountDownLatch(1);
+    final HttpClient.Builder builder = getHttpClientFactory().newBuilder()
+        .connectTimeout(1, TimeUnit.SECONDS)
+        .addOrReplaceInterceptor("test", new Interceptor() {
+          @Override
+          public void afterConnectionFailure(HttpRequest request, Throwable failure) {
+            server = new DefaultMockServer(false);
+            server.expect().withPath("/intercepted-url").andReturn(200, "This works").once();
+            server.start(originalPort); // Need to restart on the original port as we can't alter the request during retry.
+          }
+
+          @Override
+          public void after(HttpRequest request, HttpResponse<?> response, Consumer<List<ByteBuffer>> consumer) {
+            afterInvoked.countDown();
           }
         });
     // When
     try (HttpClient client = builder.build()) {
-      final CompletableFuture<HttpResponse<String>> response = client.sendAsync(client.newHttpRequestBuilder().uri(server.url("/not-found")).build(), String.class);
+      final CompletableFuture<HttpResponse<String>> response = client.sendAsync(client.newHttpRequestBuilder()
+          .timeout(1, TimeUnit.SECONDS)
+          .uri(server.url("/intercepted-url")).build(), String.class);
 
       // Then
-      assertThat(response).succeedsWithin(Duration.of(10, ChronoUnit.SECONDS));
+      assertThat(response).succeedsWithin(Duration.of(30, ChronoUnit.SECONDS));
+      assertThat(afterInvoked).extracting(CountDownLatch::getCount).isEqualTo(0L);
     }
   }