From f295a08ac0f7fac0af46d989c9aea71716dca438 Mon Sep 17 00:00:00 2001 From: bharathreddygunapati Date: Mon, 26 Oct 2026 21:03:43 +0530 Subject: [PATCH] [FLINK-41192][connector-http] Fail the HTTP sink request when a retry attempt throws Resilience4j does not catch an exception thrown while starting an attempt. On a retry attempt, which runs on the retry scheduler, the exception was lost and the request future never completed, so the sink request stayed in flight until the AsyncSink request timeout and was then resent whole. Wrap the attempt so such an exception becomes a failed stage: an I/O failure in its cause chain is retried like a transport error, anything else fails the request through the writer's existing exception path. Generated-by: Claude Code (Claude Fable 5.1) --- .../httpclient/HttpSinkClientWithRetry.java | 20 +++- .../HttpSinkClientWithRetryTest.java | 93 +++++++++++++++++++ .../httpclient/JavaNetSinkHttpClientTest.java | 26 ++++++ 3 files changed, 138 insertions(+), 1 deletion(-) diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetry.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetry.java index ef048c9e..aaed2e50 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetry.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetry.java @@ -72,7 +72,7 @@ CompletableFuture send( return Retry.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 submitAttempt( + Function, CompletionStage> + attemptSubmitter, + List requestEntries) { + try { + return attemptSubmitter.apply(requestEntries); + } catch (RuntimeException e) { + return CompletableFuture.failedFuture(e); + } + } + static boolean isRetryableIoException(Throwable throwable) { Throwable current = throwable; while (current != null) { diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetryTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetryTest.java index f17cacf0..45750420 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/HttpSinkClientWithRetryTest.java +++ b/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); + } } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java index 9fb5eae2..1f61e226 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java +++ b/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 recordingCallback( List> httpRequests) { return (response, request, endpointUrl, headerMap) ->