[FLINK-41192][connector-http] Fail the HTTP sink request when a retry attempt throws

flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetry.java

@@ -72,7 +72,7 @@ CompletableFuture<SinkHttpClientResponses> send(
         return Retry.<HttpSinkAttemptResult>decorateCompletionStage(
                         retry,
                         retryScheduler,
-                        () -> attemptSubmitter.apply(requestEntriesToSubmit.get()))
+                        () -> submitAttempt(attemptSubmitter, requestEntriesToSubmit.get()))
                 .get()
                 .thenApply(
                         attemptResult -> {
@@ -88,6 +88,24 @@ void close() {
         retryScheduler.shutdownNow();
     }
 
+    /**
+     * Runs one attempt and turns an exception thrown while starting it into a failed stage.
+     * Resilience4j does not catch exceptions thrown by the attempt itself: on a retry attempt,
+     * which runs on the retry scheduler, the exception would be lost and the request would never
+     * complete (FLINK-41192). As a failed stage it goes through {@code retryOnException}, so an I/O
+     * failure is retried like a transport error and anything else fails the request.
+     */
+    private static CompletionStage<HttpSinkAttemptResult> submitAttempt(
+            Function<List<HttpSinkRequestEntry>, CompletionStage<HttpSinkAttemptResult>>
+                    attemptSubmitter,
+            List<HttpSinkRequestEntry> requestEntries) {
+        try {
+            return attemptSubmitter.apply(requestEntries);
+        } catch (RuntimeException e) {
+            return CompletableFuture.failedFuture(e);
+        }
+    }
+
     static boolean isRetryableIoException(Throwable throwable) {
         Throwable current = throwable;
         while (current != null) {

flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetryTest.java

@@ -27,6 +27,8 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -209,4 +211,95 @@ public void testDoesNotRetryNonIoExceptionFromFailedCompletionStage() {
 
         assertThat(calls).hasValue(1);
     }
+
+    @Test
+    public void testRetryAttemptThatThrowsCompletesTheRequestExceptionally() throws Exception {
+        HttpSinkRequestEntry retryableEntry = new HttpSinkRequestEntry("POST", new byte[] {1});
+        AtomicInteger calls = new AtomicInteger();
+        HttpSinkClientWithRetry retryingClient =
+                new HttpSinkClientWithRetry(
+                        RetryConfig.custom()
+                                .maxAttempts(3)
+                                .intervalFunction(IntervalFunction.of(Duration.ofMillis(1)))
+                                .build());
+
+        var future =
+                retryingClient.send(
+                        List.of(retryableEntry),
+                        requestEntries -> {
+                            if (calls.getAndIncrement() == 0) {
+                                HttpSinkAttemptResult attemptResult = new HttpSinkAttemptResult();
+                                attemptResult.addRetryableRequests(List.of(retryableEntry));
+                                return CompletableFuture.completedFuture(attemptResult);
+                            }
+                            // e.g. the OIDC token endpoint rejects the token request while the
+                            // retry attempt builds the HTTP request on the retry scheduler
+                            throw new IllegalStateException("token endpoint returned 500");
+                        });
+
+        // must not hang: a retry attempt that throws has to fail the request
+        assertThatThrownBy(() -> future.get(5, TimeUnit.SECONDS))
+                .isInstanceOf(ExecutionException.class)
+                .hasCauseInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("token endpoint returned 500");
+        assertThat(calls).hasValue(2);
+    }
+
+    @Test
+    public void testRetryAttemptThatThrowsRetryableIoExceptionIsRetried() throws Exception {
+        HttpSinkRequestEntry retryableEntry = new HttpSinkRequestEntry("POST", new byte[] {1});
+        AtomicInteger calls = new AtomicInteger();
+        HttpSinkClientWithRetry retryingClient =
+                new HttpSinkClientWithRetry(
+                        RetryConfig.custom()
+                                .maxAttempts(3)
+                                .intervalFunction(IntervalFunction.of(Duration.ofMillis(1)))
+                                .build());
+
+        var response =
+                retryingClient
+                        .send(
+                                List.of(retryableEntry),
+                                requestEntries -> {
+                                    int call = calls.getAndIncrement();
+                                    if (call == 0) {
+                                        HttpSinkAttemptResult attemptResult =
+                                                new HttpSinkAttemptResult();
+                                        attemptResult.addRetryableRequests(List.of(retryableEntry));
+                                        return CompletableFuture.completedFuture(attemptResult);
+                                    }
+                                    if (call == 1) {
+                                        throw new IllegalStateException(
+                                                "token request failed",
+                                                new java.io.IOException("connection reset"));
+                                    }
+                                    HttpSinkAttemptResult attemptResult =
+                                            new HttpSinkAttemptResult();
+                                    attemptResult.addSuccessfulRequests(List.of(retryableEntry));
+                                    return CompletableFuture.completedFuture(attemptResult);
+                                })
+                        .get(5, TimeUnit.SECONDS);
+
+        assertThat(response.getSuccessfulRequests()).containsExactly(retryableEntry);
+        assertThat(response.getRetriableFailedRequests()).isEmpty();
+        assertThat(calls).hasValue(3);
+    }
+
+    @Test
+    public void testFirstAttemptThatThrowsCompletesTheRequestExceptionally() throws Exception {
+        HttpSinkRequestEntry entry = new HttpSinkRequestEntry("POST", new byte[] {1});
+        HttpSinkClientWithRetry retryingClient =
+                new HttpSinkClientWithRetry(RetryConfig.custom().maxAttempts(1).build());
+
+        var future =
+                retryingClient.send(
+                        List.of(entry),
+                        requestEntries -> {
+                            throw new IllegalStateException("cannot build request");
+                        });
+
+        assertThatThrownBy(() -> future.get(5, TimeUnit.SECONDS))
+                .isInstanceOf(ExecutionException.class)
+                .hasCauseInstanceOf(IllegalStateException.class);
+    }
 }

flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java

@@ -591,6 +591,32 @@ private JavaNetSinkHttpClient retryingClient(
         return client;
     }
 
+    @Test
+    public void shouldFailTheRequestWhenARetryAttemptThrows() {
+        HttpSinkRequestEntry retryableEntry = new HttpSinkRequestEntry("POST", new byte[] {1});
+        AtomicInteger submits = new AtomicInteger();
+        RequestSubmitterFactory submitterFactory =
+                (_sinkConfig, _headersAndValues) ->
+                        (_endpointUrl, _requestToSubmit) -> {
+                            if (submits.getAndIncrement() == 0) {
+                                return List.of(responseFuture(retryableEntry, 503));
+                            }
+                            throw new IllegalStateException("token endpoint returned 500");
+                        };
+
+        Properties properties = new Properties();
+        properties.setProperty(SINK_MAX_RETRIES, "2");
+        properties.setProperty(SINK_RETRY_FIXED_DELAY_DELAY, "1ms");
+        JavaNetSinkHttpClient client = retryingClient(sinkConfig(properties), submitterFactory);
+
+        var future = client.putRequests(List.of(retryableEntry), "http://localhost");
+
+        assertThatThrownBy(() -> future.get(5, java.util.concurrent.TimeUnit.SECONDS))
+                .isInstanceOf(java.util.concurrent.ExecutionException.class)
+                .hasMessageContaining("token endpoint returned 500");
+        assertThat(submits).hasValue(2);
+    }
+
     private static HttpPostRequestCallback<HttpRequest> recordingCallback(
             List<List<HttpSinkRequestEntry>> httpRequests) {
         return (response, request, endpointUrl, headerMap) ->