From 567bb0e017538176dd50c005e30c37d6b2b03d02 Mon Sep 17 00:00:00 2001 From: FeatZhang Date: Thu, 12 Mar 2026 19:11:36 +0800 Subject: [PATCH 1/3] [FLINK-39366][connector-http] Add retry support for HTTP Sink - Add SINK_HTTP_RETRY_TIMES constant to HttpConnectorConfigConstants - Add RETRY_TIMES ConfigOption (default: 3) to HttpDynamicSinkConnectorOptions - Register RETRY_TIMES as optional option in HttpDynamicTableSinkFactory - Pass retry times via Properties in HttpDynamicSink - Implement exponential backoff retry logic in HttpSinkWriter - Only retries on IOException (network errors) - Uses exponential backoff with initial delay of 1 second - Set retry.times=0 to disable retries - Update unit tests to cover retry behavior - Update documentation (English and Chinese) with new option description --- docs/content.zh/docs/connectors/table/http.md | 1 + docs/content/docs/connectors/table/http.md | 1 + .../config/HttpConnectorConfigConstants.java | 2 + .../connector/http/sink/HttpSinkWriter.java | 78 ++++++++++++------- .../http/table/sink/HttpDynamicSink.java | 5 ++ .../sink/HttpDynamicSinkConnectorOptions.java | 10 +++ .../sink/HttpDynamicTableSinkFactory.java | 2 + .../http/sink/HttpSinkWriterTest.java | 57 +++++++++++++- 8 files changed, 129 insertions(+), 27 deletions(-) diff --git a/docs/content.zh/docs/connectors/table/http.md b/docs/content.zh/docs/connectors/table/http.md index 8599e463..ff6ac579 100644 --- a/docs/content.zh/docs/connectors/table/http.md +++ b/docs/content.zh/docs/connectors/table/http.md @@ -547,6 +547,7 @@ another format name. | http.sink.writer.thread-pool.size | optional | Sets the size of pool thread for HTTP Sink request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 1 thread will be used. | | http.sink.writer.request.mode | optional | Sets the Http Sink request submission mode. Two modes are available: `single` and `batch`. Defaults to `batch` if not specified. | | http.sink.request.batch.size | optional | Applicable only for `http.sink.writer.request.mode = batch`. Sets number of individual events/requests that will be submitted as one HTTP request by HTTP sink. The default value is 500 which is same as HTTP Sink `maxBatchSize` | +| http.sink.retry.times | optional | Maximum number of retry attempts for HTTP Sink requests when a request fails due to a network error (IOException). Retries use exponential backoff with an initial delay of 1 second. Set to `0` to disable retries. Default value is `3`. | ### Sink table HTTP status codes You can configure a list of HTTP status codes that should be treated as errors for HTTP sink table. diff --git a/docs/content/docs/connectors/table/http.md b/docs/content/docs/connectors/table/http.md index 8599e463..ff6ac579 100644 --- a/docs/content/docs/connectors/table/http.md +++ b/docs/content/docs/connectors/table/http.md @@ -547,6 +547,7 @@ another format name. | http.sink.writer.thread-pool.size | optional | Sets the size of pool thread for HTTP Sink request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 1 thread will be used. | | http.sink.writer.request.mode | optional | Sets the Http Sink request submission mode. Two modes are available: `single` and `batch`. Defaults to `batch` if not specified. | | http.sink.request.batch.size | optional | Applicable only for `http.sink.writer.request.mode = batch`. Sets number of individual events/requests that will be submitted as one HTTP request by HTTP sink. The default value is 500 which is same as HTTP Sink `maxBatchSize` | +| http.sink.retry.times | optional | Maximum number of retry attempts for HTTP Sink requests when a request fails due to a network error (IOException). Retries use exponential backoff with an initial delay of 1 second. Set to `0` to disable retries. Default value is `3`. | ### Sink table HTTP status codes You can configure a list of HTTP status codes that should be treated as errors for HTTP sink table. diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java index 6dfb414e..8d88f434 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java @@ -132,6 +132,8 @@ public final class HttpConnectorConfigConstants { public static final String SINK_HTTP_WRITER_THREAD_POOL_SIZE = FLINK_CONNECTOR_HTTP + "sink.writer.thread-pool.size"; + public static final String SINK_HTTP_RETRY_TIMES = FLINK_CONNECTOR_HTTP + "sink.retry.times"; + // ----------------------------------------------------- // ------ Sink request submitter settings ------ diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java index 9513e115..52a726c8 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java @@ -36,6 +36,7 @@ import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; /** @@ -51,6 +52,10 @@ public class HttpSinkWriter extends AsyncSinkWriter extends AsyncSinkWriter elementConverter, Sink.InitContext context, @@ -96,6 +103,12 @@ public HttpSinkWriter( HttpConnectorConfigConstants.SINK_HTTP_WRITER_THREAD_POOL_SIZE, HTTP_SINK_WRITER_THREAD_POOL_SIZE)); + this.maxRetryTimes = + Integer.parseInt( + properties.getProperty( + HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES, + String.valueOf(DEFAULT_MAX_RETRY_TIMES))); + this.sinkWriterThreadPool = Executors.newFixedThreadPool( sinkWriterThreadPoolSize, @@ -103,45 +116,58 @@ public HttpSinkWriter( "http-sink-writer-worker", ThreadUtils.LOGGING_EXCEPTION_HANDLER)); } - // TODO: Reintroduce retries by adding backoff policy @Override protected void submitRequestEntries( List requestEntries, Consumer> requestResult) { + submitWithRetry(requestEntries, requestResult, 0); + } + + private void submitWithRetry( + List requestEntries, + Consumer> requestResult, + int attempt) { var future = sinkHttpClient.putRequests(requestEntries, endpointUrl); future.whenCompleteAsync( (response, err) -> { if (err != null) { - int failedRequestsNumber = requestEntries.size(); - log.error( - "Http Sink fatally failed to write all {} requests", - failedRequestsNumber); - numRecordsSendErrorsCounter.inc(failedRequestsNumber); - - // TODO: Make `HttpSinkInternal` retry the failed requests. - // Currently, it does not retry those at all, only adds their count - // to the `numRecordsSendErrors` metric. It is due to the fact we do not - // have - // a clear image how we want to do it, so it would be both efficient and - // correct. - // requestResult.accept(requestEntries); + if (attempt < maxRetryTimes) { + long backoffMs = RETRY_INITIAL_BACKOFF_MS * (1L << attempt); + log.warn( + "Http Sink failed to write {} requests due to error, " + + "retrying (attempt {}/{}) after {}ms: {}", + requestEntries.size(), + attempt + 1, + maxRetryTimes, + backoffMs, + err.getMessage()); + sinkWriterThreadPool.submit( + () -> { + try { + TimeUnit.MILLISECONDS.sleep(backoffMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + submitWithRetry(requestEntries, requestResult, attempt + 1); + }); + } else { + int failedRequestsNumber = requestEntries.size(); + log.error( + "Http Sink fatally failed to write all {} requests" + + " after {} retries", + failedRequestsNumber, + maxRetryTimes); + numRecordsSendErrorsCounter.inc(failedRequestsNumber); + requestResult.accept(Collections.emptyList()); + } } else if (response.getFailedRequests().size() > 0) { int failedRequestsNumber = response.getFailedRequests().size(); log.error("Http Sink failed to write {} requests", failedRequestsNumber); numRecordsSendErrorsCounter.inc(failedRequestsNumber); - - // TODO: Make `HttpSinkInternal` retry the failed requests. Currently, - // it does not retry those at all, only adds their count to the - // `numRecordsSendErrors` metric. It is due to the fact we do not have - // a clear image how we want to do it, so it would be both efficient and - // correct. - - // requestResult.accept(response.getFailedRequests()); - // } else { - // requestResult.accept(Collections.emptyList()); - // } + requestResult.accept(Collections.emptyList()); + } else { + requestResult.accept(Collections.emptyList()); } - requestResult.accept(Collections.emptyList()); }, sinkWriterThreadPool); } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java index 4d0e53ce..c5778bbd 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java @@ -44,7 +44,9 @@ import java.util.Properties; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.INSERT_METHOD; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.RETRY_TIMES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.URL; /** @@ -155,6 +157,9 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { .setElementConverter( new SerializationSchemaElementConverter( insertMethod, serializationSchema)) + .setProperty( + SINK_HTTP_RETRY_TIMES, + String.valueOf(tableOptions.get(RETRY_TIMES))) .setProperties(properties); addAsyncOptionsToSinkBuilder(builder); diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java index ca10cd0e..54831846 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java @@ -22,6 +22,7 @@ import java.time.Duration; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES; import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_HTTP_TIMEOUT_SECONDS; import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_REQUEST_CALLBACK_IDENTIFIER; @@ -58,4 +59,13 @@ public class HttpDynamicSinkConnectorOptions { ConfigOptions.key(SINK_REQUEST_CALLBACK_IDENTIFIER) .stringType() .defaultValue(Slf4jHttpPostRequestCallbackFactory.IDENTIFIER); + + public static final ConfigOption RETRY_TIMES = + ConfigOptions.key(SINK_HTTP_RETRY_TIMES) + .intType() + .defaultValue(3) + .withDescription( + "Maximum number of retry attempts for HTTP Sink requests on IOException. " + + "Retries use exponential backoff with an initial delay of 1 second. " + + "Set to 0 to disable retries."); } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java index 28eefae3..14827db6 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java @@ -33,6 +33,7 @@ import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.INSERT_METHOD; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.REQUEST_CALLBACK_IDENTIFIER; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.RETRY_TIMES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.URL; @@ -98,6 +99,7 @@ public Set> optionalOptions() { options.add(INSERT_METHOD); options.add(SINK_REQUEST_TIMEOUT); options.add(REQUEST_CALLBACK_IDENTIFIER); + options.add(RETRY_TIMES); return options; } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java index 865daadd..414ff861 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java @@ -44,6 +44,7 @@ import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -76,6 +77,12 @@ public void setUp() { Collection> stateBuffer = new ArrayList<>(); + Properties noRetryProperties = new Properties(); + noRetryProperties.setProperty( + org.apache.flink.connector.http.config.HttpConnectorConfigConstants + .SINK_HTTP_RETRY_TIMES, + "0"); + this.httpSinkWriter = new HttpSinkWriter<>( elementConverter, @@ -89,7 +96,7 @@ public void setUp() { "http://localhost/client", httpClient, stateBuffer, - new Properties()); + noRetryProperties); } @Test @@ -111,4 +118,52 @@ public void testErrorMetric() throws InterruptedException { Thread.sleep(2000); verify(errorCounter).inc(requestEntries.size()); } + + @Test + public void testRetryOnError() throws InterruptedException { + // default maxRetryTimes is 3, so putRequests will be called 1 + 3 = 4 times + CompletableFuture failedFuture = new CompletableFuture<>(); + failedFuture.completeExceptionally(new Exception("Connection refused")); + + when(httpClient.putRequests(anyList(), anyString())).thenReturn(failedFuture); + + Properties properties = new Properties(); + // set retry.times = 1 to speed up the test + properties.setProperty( + org.apache.flink.connector.http.config.HttpConnectorConfigConstants + .SINK_HTTP_RETRY_TIMES, + "1"); + + Collection< + org.apache.flink.connector.base.sink.writer.BufferedRequestState< + HttpSinkRequestEntry>> + stateBuffer = new ArrayList<>(); + HttpSinkWriter writerWithRetry = + new HttpSinkWriter<>( + elementConverter, + context, + 10, + 10, + 100, + 10, + 10, + 10, + "http://localhost/client", + httpClient, + stateBuffer, + properties); + + HttpSinkRequestEntry request = new HttpSinkRequestEntry("PUT", "hello".getBytes()); + Consumer> requestResult = + httpSinkRequestEntries -> log.info(String.valueOf(httpSinkRequestEntries)); + + List requestEntries = Collections.singletonList(request); + writerWithRetry.submitRequestEntries(requestEntries, requestResult); + + // Wait for 1 initial attempt + 1 retry + exponential backoff (1s) + buffer + Thread.sleep(4000); + // 1 initial attempt + 1 retry = 2 total calls + verify(httpClient, times(2)).putRequests(anyList(), anyString()); + verify(errorCounter).inc(requestEntries.size()); + } } From 286fd229864e07183d2b9a789942fca088201b70 Mon Sep 17 00:00:00 2001 From: FeatZhang Date: Tue, 31 Mar 2026 17:41:51 +0800 Subject: [PATCH 2/3] Fix retry logic issues in HttpSinkWriter - Remove premature numRecordsSendErrorsCounter.inc() before retry attempts; only count errors after all retries are exhausted - Add retry support for HTTP-level failures (getFailedRequests() non-empty), consistent with IOException retry behavior - Extract scheduleRetry() helper to reduce code duplication - Add test cases: testRetryOnHttpFailedRequests, testNoRetryWhenDisabled --- .../connector/http/sink/HttpSinkWriter.java | 54 ++++++++--- .../http/sink/HttpSinkWriterTest.java | 91 +++++++++++++++++++ 2 files changed, 132 insertions(+), 13 deletions(-) diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java index 52a726c8..2822f295 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java @@ -131,6 +131,7 @@ private void submitWithRetry( future.whenCompleteAsync( (response, err) -> { if (err != null) { + // Network-level failure (e.g. IOException) if (attempt < maxRetryTimes) { long backoffMs = RETRY_INITIAL_BACKOFF_MS * (1L << attempt); log.warn( @@ -141,15 +142,7 @@ private void submitWithRetry( maxRetryTimes, backoffMs, err.getMessage()); - sinkWriterThreadPool.submit( - () -> { - try { - TimeUnit.MILLISECONDS.sleep(backoffMs); - } catch (InterruptedException ie) { - Thread.currentThread().interrupt(); - } - submitWithRetry(requestEntries, requestResult, attempt + 1); - }); + scheduleRetry(requestEntries, requestResult, attempt, backoffMs); } else { int failedRequestsNumber = requestEntries.size(); log.error( @@ -160,18 +153,53 @@ private void submitWithRetry( numRecordsSendErrorsCounter.inc(failedRequestsNumber); requestResult.accept(Collections.emptyList()); } - } else if (response.getFailedRequests().size() > 0) { + } else if (!response.getFailedRequests().isEmpty()) { + // HTTP-level failure (e.g. 5xx response) int failedRequestsNumber = response.getFailedRequests().size(); - log.error("Http Sink failed to write {} requests", failedRequestsNumber); - numRecordsSendErrorsCounter.inc(failedRequestsNumber); - requestResult.accept(Collections.emptyList()); + if (attempt < maxRetryTimes) { + long backoffMs = RETRY_INITIAL_BACKOFF_MS * (1L << attempt); + log.warn( + "Http Sink received {} failed HTTP responses, " + + "retrying (attempt {}/{}) after {}ms", + failedRequestsNumber, + attempt + 1, + maxRetryTimes, + backoffMs); + // Retry with the original requestEntries since HttpRequest is an + // internal representation and cannot be passed back to putRequests + scheduleRetry(requestEntries, requestResult, attempt, backoffMs); + } else { + log.error( + "Http Sink failed to write {} requests after {} retries", + failedRequestsNumber, + maxRetryTimes); + numRecordsSendErrorsCounter.inc(failedRequestsNumber); + requestResult.accept(Collections.emptyList()); + } } else { + // All requests succeeded requestResult.accept(Collections.emptyList()); } }, sinkWriterThreadPool); } + private void scheduleRetry( + List requestEntries, + Consumer> requestResult, + int attempt, + long backoffMs) { + sinkWriterThreadPool.submit( + () -> { + try { + TimeUnit.MILLISECONDS.sleep(backoffMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + submitWithRetry(requestEntries, requestResult, attempt + 1); + }); + } + @Override protected long getSizeInBytes(HttpSinkRequestEntry s) { return s.getSizeInBytes(); diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java index 414ff861..6db7838a 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java @@ -23,6 +23,7 @@ import org.apache.flink.connector.base.sink.writer.ElementConverter; import org.apache.flink.connector.http.clients.SinkHttpClient; import org.apache.flink.connector.http.clients.SinkHttpClientResponse; +import org.apache.flink.connector.http.sink.httpclient.HttpRequest; import org.apache.flink.metrics.Counter; import org.apache.flink.metrics.groups.OperatorIOMetricGroup; import org.apache.flink.metrics.groups.SinkWriterMetricGroup; @@ -40,8 +41,10 @@ import java.util.List; import java.util.Properties; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.times; @@ -166,4 +169,92 @@ public void testRetryOnError() throws InterruptedException { verify(httpClient, times(2)).putRequests(anyList(), anyString()); verify(errorCounter).inc(requestEntries.size()); } + + @Test + public void testRetryOnHttpFailedRequests() throws InterruptedException { + // Simulate HTTP-level failure: server returns failed requests in response + HttpSinkRequestEntry request = new HttpSinkRequestEntry("PUT", "hello".getBytes()); + List requestEntries = Collections.singletonList(request); + + // Build a mock HttpRequest to put in the failed list + HttpRequest mockHttpRequest = + new HttpRequest(null, Collections.singletonList("hello".getBytes()), "PUT"); + + // First call: returns HTTP-level failure, second call: returns success + SinkHttpClientResponse failedResponse = + new SinkHttpClientResponse( + Collections.emptyList(), Collections.singletonList(mockHttpRequest)); + SinkHttpClientResponse successResponse = + new SinkHttpClientResponse( + Collections.singletonList(mockHttpRequest), Collections.emptyList()); + + when(httpClient.putRequests(anyList(), anyString())) + .thenReturn(CompletableFuture.completedFuture(failedResponse)) + .thenReturn(CompletableFuture.completedFuture(successResponse)); + + Properties properties = new Properties(); + properties.setProperty( + org.apache.flink.connector.http.config.HttpConnectorConfigConstants + .SINK_HTTP_RETRY_TIMES, + "2"); + + Collection> stateBuffer = new ArrayList<>(); + HttpSinkWriter writerWithRetry = + new HttpSinkWriter<>( + elementConverter, + context, + 10, + 10, + 100, + 10, + 10, + 10, + "http://localhost/client", + httpClient, + stateBuffer, + properties); + + AtomicInteger acceptCallCount = new AtomicInteger(0); + Consumer> requestResult = + httpSinkRequestEntries -> acceptCallCount.incrementAndGet(); + + writerWithRetry.submitRequestEntries(requestEntries, requestResult); + + // Wait for retry backoff (1s) + buffer + Thread.sleep(3000); + + // Should have retried once and then succeeded: 2 total calls + verify(httpClient, times(2)).putRequests(anyList(), anyString()); + // No error counted since eventually succeeded + verify(errorCounter, times(0)).inc(requestEntries.size()); + // requestResult.accept() called exactly once + assertThat(acceptCallCount.get()).isEqualTo(1); + } + + @Test + public void testNoRetryWhenDisabled() throws InterruptedException { + // retry.times=0 should not retry at all (already covered by setUp, explicit test here) + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new Exception("Connection refused")); + + when(httpClient.putRequests(anyList(), anyString())).thenReturn(future); + + HttpSinkRequestEntry request = new HttpSinkRequestEntry("PUT", "hello".getBytes()); + List requestEntries = Collections.singletonList(request); + + AtomicInteger acceptCallCount = new AtomicInteger(0); + Consumer> requestResult = + httpSinkRequestEntries -> acceptCallCount.incrementAndGet(); + + // httpSinkWriter is created with retry.times=0 in setUp + this.httpSinkWriter.submitRequestEntries(requestEntries, requestResult); + + Thread.sleep(1000); + + // Only 1 attempt, no retries + verify(httpClient, times(1)).putRequests(anyList(), anyString()); + verify(errorCounter).inc(requestEntries.size()); + // requestResult.accept() called exactly once + assertThat(acceptCallCount.get()).isEqualTo(1); + } } From 82b9450c88902c4b19d31bcab61704d94ec95825 Mon Sep 17 00:00:00 2001 From: FeatZhang Date: Fri, 8 May 2026 15:17:09 +0800 Subject: [PATCH 3/3] [FLINK-39366][connector-http] Align sink retry configuration with lookup source Extend the HTTP sink retry model so it matches the lookup source feature set (Option 1 in PR review): - Add http.sink.retry-strategy.type (fixed-delay | exponential-delay). - Add http.sink.retry-codes and http.sink.success-codes so users can precisely define which HTTP status codes are transient vs fatal. - Add fixed-delay / exponential-delay tuning keys (initial-backoff, max-backoff, backoff-multiplier, delay) mirroring the ones the lookup source already exposes. - Reuse the existing retry utilities: HttpResponseChecker drives the status-code classification, resilience4j's IntervalFunction computes the backoff, and RetryConfigProvider is generalised through a RetryOptionKeys bag so sink and lookup share the same strategy logic. - SinkHttpClientResponse now distinguishes retryable (transient) and fatal failures; HttpSinkWriter retries only the retryable ones and counts fatal failures against numRecordsSendErrorsCounter immediately. - The legacy http.sink.error.code* options remain untouched: when the new retry-codes / success-codes options are not set the client falls back to the previous one-bucket behaviour for backwards compatibility. - Update both the English and the Chinese connector docs with the new options and a dedicated 'Retries and handling errors (Sink)' section. - Add SinkRetryConfigTest and a new HttpSinkWriterTest case for fatal failures. --- docs/content.zh/docs/connectors/table/http.md | 9 +- docs/content/docs/connectors/table/http.md | 22 +- .../http/clients/SinkHttpClientResponse.java | 53 ++++- .../config/HttpConnectorConfigConstants.java | 27 +++ .../http/retry/RetryConfigProvider.java | 116 ++++++++-- .../connector/http/sink/HttpSinkWriter.java | 134 +++++++----- .../connector/http/sink/SinkRetryConfig.java | 205 ++++++++++++++++++ .../httpclient/JavaNetSinkHttpClient.java | 65 +++++- .../http/table/sink/HttpDynamicSink.java | 37 ++++ .../sink/HttpDynamicSinkConnectorOptions.java | 66 +++++- .../sink/HttpDynamicTableSinkFactory.java | 14 ++ .../http/sink/HttpSinkWriterTest.java | 54 +++++ .../http/sink/SinkRetryConfigTest.java | 128 +++++++++++ 13 files changed, 836 insertions(+), 94 deletions(-) create mode 100644 flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/SinkRetryConfig.java create mode 100644 flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/SinkRetryConfigTest.java diff --git a/docs/content.zh/docs/connectors/table/http.md b/docs/content.zh/docs/connectors/table/http.md index ff6ac579..eedb3b6e 100644 --- a/docs/content.zh/docs/connectors/table/http.md +++ b/docs/content.zh/docs/connectors/table/http.md @@ -547,7 +547,14 @@ another format name. | http.sink.writer.thread-pool.size | optional | Sets the size of pool thread for HTTP Sink request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 1 thread will be used. | | http.sink.writer.request.mode | optional | Sets the Http Sink request submission mode. Two modes are available: `single` and `batch`. Defaults to `batch` if not specified. | | http.sink.request.batch.size | optional | Applicable only for `http.sink.writer.request.mode = batch`. Sets number of individual events/requests that will be submitted as one HTTP request by HTTP sink. The default value is 500 which is same as HTTP Sink `maxBatchSize` | -| http.sink.retry.times | optional | Maximum number of retry attempts for HTTP Sink requests when a request fails due to a network error (IOException). Retries use exponential backoff with an initial delay of 1 second. Set to `0` to disable retries. Default value is `3`. | +| http.sink.retry.times | optional | Maximum number of retry attempts for HTTP Sink requests when a request fails due to a network error (IOException) or when the response matches a retry code (see `http.sink.retry-codes`). Set to `0` to disable retries. Default value is `3`. | +| http.sink.retry-strategy.type | optional | Auto retry strategy type for the HTTP sink: `fixed-delay` or `exponential-delay`. Defaults to `exponential-delay`. | +| http.sink.retry-codes | optional | Comma separated http codes considered as transient errors that should be retried. Use `[1-5]XX` for groups and `!` for excluding. Default `500,503,504`. | +| http.sink.success-codes | optional | Comma separated http codes considered as a successful sink response. Use `[1-5]XX` for groups and `!` for excluding. Default `2XX`. Any status code that is neither a success nor a retry code is considered a fatal failure and is not retried. | +| http.sink.retry-strategy.fixed-delay.delay | optional | Fixed-delay interval between sink retries when `http.sink.retry-strategy.type=fixed-delay`. Default 1 second. | +| http.sink.retry-strategy.exponential-delay.initial-backoff | optional | Exponential-delay initial delay when `http.sink.retry-strategy.type=exponential-delay`. Default 1 second. | +| http.sink.retry-strategy.exponential-delay.max-backoff | optional | Exponential-delay maximum delay when `http.sink.retry-strategy.type=exponential-delay`. Default 1 minute. | +| http.sink.retry-strategy.exponential-delay.backoff-multiplier | optional | Exponential-delay multiplier when `http.sink.retry-strategy.type=exponential-delay`. Default 2.0. | ### Sink table HTTP status codes You can configure a list of HTTP status codes that should be treated as errors for HTTP sink table. diff --git a/docs/content/docs/connectors/table/http.md b/docs/content/docs/connectors/table/http.md index ff6ac579..d45d8f2c 100644 --- a/docs/content/docs/connectors/table/http.md +++ b/docs/content/docs/connectors/table/http.md @@ -547,7 +547,27 @@ another format name. | http.sink.writer.thread-pool.size | optional | Sets the size of pool thread for HTTP Sink request processing. Increasing this value would mean that more concurrent requests can be processed in the same time. If not specified, the default value of 1 thread will be used. | | http.sink.writer.request.mode | optional | Sets the Http Sink request submission mode. Two modes are available: `single` and `batch`. Defaults to `batch` if not specified. | | http.sink.request.batch.size | optional | Applicable only for `http.sink.writer.request.mode = batch`. Sets number of individual events/requests that will be submitted as one HTTP request by HTTP sink. The default value is 500 which is same as HTTP Sink `maxBatchSize` | -| http.sink.retry.times | optional | Maximum number of retry attempts for HTTP Sink requests when a request fails due to a network error (IOException). Retries use exponential backoff with an initial delay of 1 second. Set to `0` to disable retries. Default value is `3`. | +| http.sink.retry.times | optional | Maximum number of retry attempts for HTTP Sink requests when a request fails due to a network error (IOException) or when the response matches a retry code (see `http.sink.retry-codes`). Set to `0` to disable retries. Default value is `3`. | +| http.sink.retry-strategy.type | optional | Auto retry strategy type for the HTTP sink: `fixed-delay` or `exponential-delay`. Defaults to `exponential-delay`. | +| http.sink.retry-codes | optional | Comma separated http codes considered as transient errors that should be retried. Use `[1-5]XX` for groups and `!` for excluding. Default `500,503,504`. | +| http.sink.success-codes | optional | Comma separated http codes considered as a successful sink response. Use `[1-5]XX` for groups and `!` for excluding. Default `2XX`. Any status code that is neither a success nor a retry code is considered a fatal failure and is not retried. | +| http.sink.retry-strategy.fixed-delay.delay | optional | Fixed-delay interval between sink retries when `http.sink.retry-strategy.type=fixed-delay`. Default 1 second. | +| http.sink.retry-strategy.exponential-delay.initial-backoff | optional | Exponential-delay initial delay when `http.sink.retry-strategy.type=exponential-delay`. Default 1 second. | +| http.sink.retry-strategy.exponential-delay.max-backoff | optional | Exponential-delay maximum delay when `http.sink.retry-strategy.type=exponential-delay`. Default 1 minute. | +| http.sink.retry-strategy.exponential-delay.backoff-multiplier | optional | Exponential-delay multiplier when `http.sink.retry-strategy.type=exponential-delay`. Default 2.0. | + +### Retries and handling errors (Sink) +The HTTP sink uses the same retry model as the lookup source (see the `http.source.lookup.*` counterparts above): + +- Network-level failures (e.g. `IOException`) are always retryable. +- HTTP responses are classified by status code: + - success codes (`http.sink.success-codes`, default `2XX`) are treated as success and acknowledged; + - retry codes (`http.sink.retry-codes`, default `500,503,504`) trigger a retry respecting `http.sink.retry.times`; + - any other status code is a **fatal** failure — it is counted against `numRecordsSendErrorsCounter` immediately and the affected records are skipped (not retried and not blocking the pipeline). + +Users can choose a retry strategy type for the sink: +- `fixed-delay` — request will be re-sent after `http.sink.retry-strategy.fixed-delay.delay`. +- `exponential-delay` (default) — request will be re-sent with exponential backoff, limited by `http.sink.retry.times` attempts. The delay for each retry is the previous delay multiplied by `http.sink.retry-strategy.exponential-delay.backoff-multiplier`, capped at `http.sink.retry-strategy.exponential-delay.max-backoff`. The initial delay is `http.sink.retry-strategy.exponential-delay.initial-backoff`. ### Sink table HTTP status codes You can configure a list of HTTP status codes that should be treated as errors for HTTP sink table. diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/clients/SinkHttpClientResponse.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/clients/SinkHttpClientResponse.java index 9fbf1695..74bda6c3 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/clients/SinkHttpClientResponse.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/clients/SinkHttpClientResponse.java @@ -20,23 +20,66 @@ import org.apache.flink.connector.http.sink.HttpSinkRequestEntry; import org.apache.flink.connector.http.sink.httpclient.HttpRequest; -import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.Getter; import lombok.NonNull; import lombok.ToString; +import java.util.ArrayList; +import java.util.Collections; import java.util.List; /** * Data class holding {@link HttpSinkRequestEntry} instances that {@link SinkHttpClient} attempted - * to write, divided into two lists — successful and failed ones. + * to write, divided into successful, retryable (transient HTTP failures) and fatal failures. + * + *

Retryable failures are requests whose response status code matches the configured {@code + * retry-codes} (by default {@code 500,503,504}). Fatal failures are everything else that is not a + * success (4xx, non-listed 5xx, etc). The HTTP sink will only replay retryable failures; fatal + * failures are counted as errors immediately without blocking the pipeline. */ -@Data +@Getter @ToString +@EqualsAndHashCode public class SinkHttpClientResponse { /** A list of successfully written requests. */ @NonNull private final List successfulRequests; - /** A list of requests that {@link SinkHttpClient} failed to write. */ - @NonNull private final List failedRequests; + /** Requests that failed with a transient HTTP status code and may be retried. */ + @NonNull private final List retryableFailedRequests; + + /** Requests that failed with a non-retryable status code (fatal failures). */ + @NonNull private final List fatalFailedRequests; + + public SinkHttpClientResponse( + @NonNull List successfulRequests, + @NonNull List retryableFailedRequests, + @NonNull List fatalFailedRequests) { + this.successfulRequests = successfulRequests; + this.retryableFailedRequests = retryableFailedRequests; + this.fatalFailedRequests = fatalFailedRequests; + } + + /** + * Backwards compatible constructor: every failed request is considered retryable. Provided so + * existing callers / tests keep working. + */ + public SinkHttpClientResponse( + @NonNull List successfulRequests, + @NonNull List failedRequests) { + this(successfulRequests, failedRequests, Collections.emptyList()); + } + + /** + * All failed requests, regardless of whether they are retryable or fatal. Kept for backwards + * compatibility with code written before the retryable/fatal split was introduced. + */ + public List getFailedRequests() { + List all = + new ArrayList<>(retryableFailedRequests.size() + fatalFailedRequests.size()); + all.addAll(retryableFailedRequests); + all.addAll(fatalFailedRequests); + return all; + } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java index 8d88f434..3ea92c88 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/config/HttpConnectorConfigConstants.java @@ -166,4 +166,31 @@ public final class HttpConnectorConfigConstants { SOURCE_RETRY_EXP_DELAY_PREFIX + "max-backoff"; public static final String SOURCE_RETRY_EXP_DELAY_MULTIPLIER = SOURCE_RETRY_EXP_DELAY_PREFIX + "backoff-multiplier"; + + // ---------------- Sink retry configuration ---------------- + // Mirrors the lookup source retry options (prefixed with http.source.lookup.*) so the + // sink side offers the same level of control. All sink retry keys are prefixed with + // http.sink.* and can be consumed through the Table API DDL or via the DataStream + // HttpSinkBuilder's properties. + private static final String SINK_PREFIX = FLINK_CONNECTOR_HTTP + "sink."; + + public static final String SINK_RETRY_SUCCESS_CODES = SINK_PREFIX + "success-codes"; + public static final String SINK_RETRY_RETRY_CODES = SINK_PREFIX + "retry-codes"; + + public static final String SINK_RETRY_STRATEGY_PREFIX = SINK_PREFIX + "retry-strategy."; + public static final String SINK_RETRY_STRATEGY_TYPE = SINK_RETRY_STRATEGY_PREFIX + "type"; + + private static final String SINK_RETRY_FIXED_DELAY_PREFIX = + SINK_RETRY_STRATEGY_PREFIX + "fixed-delay."; + public static final String SINK_RETRY_FIXED_DELAY_DELAY = + SINK_RETRY_FIXED_DELAY_PREFIX + "delay"; + + private static final String SINK_RETRY_EXP_DELAY_PREFIX = + SINK_RETRY_STRATEGY_PREFIX + "exponential-delay."; + public static final String SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF = + SINK_RETRY_EXP_DELAY_PREFIX + "initial-backoff"; + public static final String SINK_RETRY_EXP_DELAY_MAX_BACKOFF = + SINK_RETRY_EXP_DELAY_PREFIX + "max-backoff"; + public static final String SINK_RETRY_EXP_DELAY_MULTIPLIER = + SINK_RETRY_EXP_DELAY_PREFIX + "backoff-multiplier"; } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/retry/RetryConfigProvider.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/retry/RetryConfigProvider.java index 424113bb..4d7838c1 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/retry/RetryConfigProvider.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/retry/RetryConfigProvider.java @@ -16,6 +16,7 @@ package org.apache.flink.connector.http.retry; +import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.table.connector.source.lookup.LookupOptions; @@ -24,6 +25,8 @@ import lombok.AccessLevel; import lombok.RequiredArgsConstructor; +import java.time.Duration; + import static io.github.resilience4j.core.IntervalFunction.ofExponentialBackoff; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF; @@ -31,45 +34,116 @@ import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_FIXED_DELAY_DELAY; import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_STRATEGY; -/** Configuration for Retry. */ +/** + * Configuration for Retry. + * + *

The provider is generic: it works for the lookup source (via {@link #create(ReadableConfig)}) + * and for any other component (e.g. the HTTP sink) by passing a custom set of {@link ConfigOption}s + * through {@link #create(ReadableConfig, RetryOptionKeys)}. + */ @RequiredArgsConstructor(access = AccessLevel.PRIVATE) public class RetryConfigProvider { private final ReadableConfig config; + private final RetryOptionKeys keys; + private final int maxAttempts; + /** Create a {@link RetryConfig} using the lookup-source defaults and {@code max-retries}. */ public static RetryConfig create(ReadableConfig config) { - return new RetryConfigProvider(config).create(); + return new RetryConfigProvider( + config, + RetryOptionKeys.lookupSource(), + config.get(LookupOptions.MAX_RETRIES) + 1) + .build(); + } + + /** + * Create a {@link RetryConfig} for a component providing its own option keys and a pre-computed + * max-attempts value (max-retries + 1). + */ + public static RetryConfig create(ReadableConfig config, RetryOptionKeys keys, int maxAttempts) { + return new RetryConfigProvider(config, keys, maxAttempts).build(); + } + + /** + * Build the {@link IntervalFunction} alone — handy for components that drive their own retry + * loop (e.g. {@code HttpSinkWriter}) but still want to share the fixed-delay / + * exponential-delay behaviour with the lookup source. + */ + public static IntervalFunction intervalFunction(ReadableConfig config, RetryOptionKeys keys) { + return new RetryConfigProvider(config, keys, 1).buildIntervalFunction(); } - private RetryConfig create() { - return createBuilder().maxAttempts(config.get(LookupOptions.MAX_RETRIES) + 1).build(); + private RetryConfig build() { + return createBuilder().maxAttempts(maxAttempts).build(); } private RetryConfig.Builder createBuilder() { - var retryStrategy = getRetryStrategy(); + return RetryConfig.custom().intervalFunction(buildIntervalFunction()); + } + + private IntervalFunction buildIntervalFunction() { + var retryStrategy = RetryStrategyType.fromCode(config.get(keys.strategy())); if (retryStrategy == RetryStrategyType.FIXED_DELAY) { - return configureFixedDelay(); + return IntervalFunction.of(config.get(keys.fixedDelay())); } else if (retryStrategy == RetryStrategyType.EXPONENTIAL_DELAY) { - return configureExponentialDelay(); + Duration initialDelay = config.get(keys.exponentialInitialBackoff()); + Duration maxDelay = config.get(keys.exponentialMaxBackoff()); + double multiplier = config.get(keys.exponentialMultiplier()); + return ofExponentialBackoff(initialDelay, multiplier, maxDelay); } throw new IllegalArgumentException("Unsupported retry strategy: " + retryStrategy); } - private RetryStrategyType getRetryStrategy() { - return RetryStrategyType.fromCode(config.get(SOURCE_LOOKUP_RETRY_STRATEGY)); - } + /** Bag of {@link ConfigOption} references identifying the retry-related options. */ + public static final class RetryOptionKeys { - private RetryConfig.Builder configureFixedDelay() { - return RetryConfig.custom() - .intervalFunction( - IntervalFunction.of(config.get(SOURCE_LOOKUP_RETRY_FIXED_DELAY_DELAY))); - } + private final ConfigOption strategy; + private final ConfigOption fixedDelay; + private final ConfigOption exponentialInitialBackoff; + private final ConfigOption exponentialMaxBackoff; + private final ConfigOption exponentialMultiplier; + + public RetryOptionKeys( + ConfigOption strategy, + ConfigOption fixedDelay, + ConfigOption exponentialInitialBackoff, + ConfigOption exponentialMaxBackoff, + ConfigOption exponentialMultiplier) { + this.strategy = strategy; + this.fixedDelay = fixedDelay; + this.exponentialInitialBackoff = exponentialInitialBackoff; + this.exponentialMaxBackoff = exponentialMaxBackoff; + this.exponentialMultiplier = exponentialMultiplier; + } + + static RetryOptionKeys lookupSource() { + return new RetryOptionKeys( + SOURCE_LOOKUP_RETRY_STRATEGY, + SOURCE_LOOKUP_RETRY_FIXED_DELAY_DELAY, + SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF, + SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF, + SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MULTIPLIER); + } + + ConfigOption strategy() { + return strategy; + } + + ConfigOption fixedDelay() { + return fixedDelay; + } + + ConfigOption exponentialInitialBackoff() { + return exponentialInitialBackoff; + } - private RetryConfig.Builder configureExponentialDelay() { - var initialDelay = config.get(SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF); - var maxDelay = config.get(SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF); - var multiplier = config.get(SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MULTIPLIER); - return RetryConfig.custom() - .intervalFunction(ofExponentialBackoff(initialDelay, multiplier, maxDelay)); + ConfigOption exponentialMaxBackoff() { + return exponentialMaxBackoff; + } + + ConfigOption exponentialMultiplier() { + return exponentialMultiplier; + } } } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java index 2822f295..773b771a 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java @@ -45,6 +45,20 @@ *

More details on the internals of this sink writer may be found in {@link AsyncSinkWriter} * documentation. * + *

The writer's retry policy mirrors the lookup source: + * + *

    + *
  • {@code http.sink.retry.times} — maximum number of retry attempts. + *
  • {@code http.sink.retry-strategy.type} — {@code fixed-delay} or {@code exponential-delay}. + *
  • {@code http.sink.retry-strategy.fixed-delay.delay} — fixed interval between retries. + *
  • {@code http.sink.retry-strategy.exponential-delay.initial-backoff / max-backoff / + * backoff-multiplier} — exponential backoff parameters. + *
  • {@code http.sink.retry-codes} — HTTP status codes that should trigger a retry (default + * {@code 500,503,504}). + *
  • {@code http.sink.success-codes} — HTTP status codes that should be treated as success + * (default {@code 2XX}). + *
+ * * @param type of the elements that should be sent through HTTP request. */ @Slf4j @@ -52,10 +66,6 @@ public class HttpSinkWriter extends AsyncSinkWriter extends AsyncSinkWriter elementConverter, @@ -103,11 +113,7 @@ public HttpSinkWriter( HttpConnectorConfigConstants.SINK_HTTP_WRITER_THREAD_POOL_SIZE, HTTP_SINK_WRITER_THREAD_POOL_SIZE)); - this.maxRetryTimes = - Integer.parseInt( - properties.getProperty( - HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES, - String.valueOf(DEFAULT_MAX_RETRY_TIMES))); + this.retryConfig = SinkRetryConfig.fromProperties(properties); this.sinkWriterThreadPool = Executors.newFixedThreadPool( @@ -131,59 +137,73 @@ private void submitWithRetry( future.whenCompleteAsync( (response, err) -> { if (err != null) { - // Network-level failure (e.g. IOException) - if (attempt < maxRetryTimes) { - long backoffMs = RETRY_INITIAL_BACKOFF_MS * (1L << attempt); - log.warn( - "Http Sink failed to write {} requests due to error, " - + "retrying (attempt {}/{}) after {}ms: {}", - requestEntries.size(), - attempt + 1, - maxRetryTimes, - backoffMs, - err.getMessage()); - scheduleRetry(requestEntries, requestResult, attempt, backoffMs); - } else { - int failedRequestsNumber = requestEntries.size(); - log.error( - "Http Sink fatally failed to write all {} requests" - + " after {} retries", - failedRequestsNumber, - maxRetryTimes); - numRecordsSendErrorsCounter.inc(failedRequestsNumber); - requestResult.accept(Collections.emptyList()); - } - } else if (!response.getFailedRequests().isEmpty()) { - // HTTP-level failure (e.g. 5xx response) - int failedRequestsNumber = response.getFailedRequests().size(); - if (attempt < maxRetryTimes) { - long backoffMs = RETRY_INITIAL_BACKOFF_MS * (1L << attempt); - log.warn( - "Http Sink received {} failed HTTP responses, " - + "retrying (attempt {}/{}) after {}ms", - failedRequestsNumber, - attempt + 1, - maxRetryTimes, - backoffMs); - // Retry with the original requestEntries since HttpRequest is an - // internal representation and cannot be passed back to putRequests - scheduleRetry(requestEntries, requestResult, attempt, backoffMs); - } else { - log.error( - "Http Sink failed to write {} requests after {} retries", - failedRequestsNumber, - maxRetryTimes); - numRecordsSendErrorsCounter.inc(failedRequestsNumber); - requestResult.accept(Collections.emptyList()); - } - } else { - // All requests succeeded + // Network-level failure (e.g. IOException). + handleRetry(requestEntries, requestResult, attempt, err.getMessage()); + return; + } + + List retryable = + response.getRetryableFailedRequests(); + List fatal = + response.getFatalFailedRequests(); + + if (!fatal.isEmpty()) { + // Non-retryable errors: record them and do not replay. + log.error( + "Http Sink received {} fatal (non-retryable) HTTP responses," + + " skipping retry", + fatal.size()); + numRecordsSendErrorsCounter.inc(fatal.size()); + } + + if (retryable.isEmpty()) { + // Either everything succeeded or only fatal failures were returned. requestResult.accept(Collections.emptyList()); + return; } + + // We have retryable failures — retry the original requestEntries because + // HttpRequest is an internal representation and cannot be passed back to + // putRequests(). Submitting the full batch keeps the behaviour consistent + // with the network-level retry path below. + handleRetry( + requestEntries, + requestResult, + attempt, + retryable.size() + " retryable HTTP failures"); }, sinkWriterThreadPool); } + private void handleRetry( + List requestEntries, + Consumer> requestResult, + int attempt, + String reason) { + int maxRetries = retryConfig.getMaxRetries(); + if (attempt < maxRetries) { + long backoffMs = retryConfig.backoffMillis(attempt); + log.warn( + "Http Sink failed to write {} requests ({}), retrying (attempt {}/{}) after" + + " {}ms", + requestEntries.size(), + reason, + attempt + 1, + maxRetries, + backoffMs); + scheduleRetry(requestEntries, requestResult, attempt, backoffMs); + } else { + int failedRequestsNumber = requestEntries.size(); + log.error( + "Http Sink fatally failed to write all {} requests after {} retries ({})", + failedRequestsNumber, + maxRetries, + reason); + numRecordsSendErrorsCounter.inc(failedRequestsNumber); + requestResult.accept(Collections.emptyList()); + } + } + private void scheduleRetry( List requestEntries, Consumer> requestResult, diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/SinkRetryConfig.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/SinkRetryConfig.java new file mode 100644 index 00000000..660e4e20 --- /dev/null +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/SinkRetryConfig.java @@ -0,0 +1,205 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.http.sink; + +import org.apache.flink.connector.http.config.HttpConnectorConfigConstants; +import org.apache.flink.connector.http.retry.RetryStrategyType; +import org.apache.flink.connector.http.status.HttpCodesParser; +import org.apache.flink.connector.http.status.HttpResponseChecker; +import org.apache.flink.util.ConfigurationException; +import org.apache.flink.util.TimeUtils; + +import io.github.resilience4j.core.IntervalFunction; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; + +import java.time.Duration; +import java.util.Properties; + +/** + * Parsed retry configuration for the HTTP sink. + * + *

Mirrors the lookup source retry configuration (strategy type, retry codes, success codes, + * fixed / exponential delay parameters) and is consumed from the sink-builder {@link Properties} + * bag so that both the Table API and the DataStream {@code HttpSinkBuilder} can share the same + * knobs. + */ +@Slf4j +@Getter +final class SinkRetryConfig { + + static final String DEFAULT_STRATEGY = RetryStrategyType.EXPONENTIAL_DELAY.getCode(); + static final String DEFAULT_SUCCESS_CODES = "2XX"; + static final String DEFAULT_RETRY_CODES = "500,503,504"; + static final Duration DEFAULT_FIXED_DELAY = Duration.ofSeconds(1); + static final Duration DEFAULT_EXP_INITIAL_BACKOFF = Duration.ofSeconds(1); + static final Duration DEFAULT_EXP_MAX_BACKOFF = Duration.ofMinutes(1); + static final double DEFAULT_EXP_MULTIPLIER = 2.0d; + static final int DEFAULT_MAX_RETRIES = 3; + + private final int maxRetries; + private final HttpResponseChecker responseChecker; + private final IntervalFunction intervalFunction; + + private SinkRetryConfig( + int maxRetries, + HttpResponseChecker responseChecker, + IntervalFunction intervalFunction) { + this.maxRetries = maxRetries; + this.responseChecker = responseChecker; + this.intervalFunction = intervalFunction; + } + + /** + * Compute the backoff (in milliseconds) for a zero-based {@code attempt} index (0 is the first + * retry, 1 is the second, ...). resilience4j's {@link IntervalFunction} is 1-based internally, + * hence the {@code +1}. + */ + long backoffMillis(int attempt) { + // IntervalFunction is 1-based: attempt=0 ⇒ apply(1) gives the first retry delay, etc. + return intervalFunction.apply(attempt + 1); + } + + /** Whether the given HTTP status code should trigger a retry. */ + boolean shouldRetry(int statusCode) { + return responseChecker.isTemporalError(statusCode); + } + + /** Whether the given HTTP status code represents a success. */ + boolean isSuccess(int statusCode) { + return responseChecker.isSuccessful(statusCode); + } + + static SinkRetryConfig fromProperties(Properties properties) { + int maxRetries = + getInt( + properties, + HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES, + DEFAULT_MAX_RETRIES); + HttpResponseChecker checker = buildChecker(properties); + IntervalFunction intervalFunction = buildIntervalFunction(properties); + return new SinkRetryConfig(maxRetries, checker, intervalFunction); + } + + private static HttpResponseChecker buildChecker(Properties properties) { + String successCodes = + properties.getProperty( + HttpConnectorConfigConstants.SINK_RETRY_SUCCESS_CODES, + DEFAULT_SUCCESS_CODES); + String retryCodes = + properties.getProperty( + HttpConnectorConfigConstants.SINK_RETRY_RETRY_CODES, DEFAULT_RETRY_CODES); + try { + return new HttpResponseChecker( + HttpCodesParser.parse(successCodes), HttpCodesParser.parse(retryCodes)); + } catch (ConfigurationException e) { + throw new IllegalArgumentException( + "Invalid HTTP sink retry codes configuration: " + e.getMessage(), e); + } + } + + private static IntervalFunction buildIntervalFunction(Properties properties) { + String rawStrategy = + properties.getProperty( + HttpConnectorConfigConstants.SINK_RETRY_STRATEGY_TYPE, DEFAULT_STRATEGY); + RetryStrategyType strategy = RetryStrategyType.fromCode(rawStrategy); + switch (strategy) { + case FIXED_DELAY: + { + Duration delay = + getDuration( + properties, + HttpConnectorConfigConstants.SINK_RETRY_FIXED_DELAY_DELAY, + DEFAULT_FIXED_DELAY); + return IntervalFunction.of(delay); + } + case EXPONENTIAL_DELAY: + { + Duration initial = + getDuration( + properties, + HttpConnectorConfigConstants + .SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF, + DEFAULT_EXP_INITIAL_BACKOFF); + Duration max = + getDuration( + properties, + HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MAX_BACKOFF, + DEFAULT_EXP_MAX_BACKOFF); + double multiplier = + getDouble( + properties, + HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MULTIPLIER, + DEFAULT_EXP_MULTIPLIER); + return IntervalFunction.ofExponentialBackoff(initial, multiplier, max); + } + default: + throw new IllegalArgumentException("Unsupported retry strategy: " + strategy); + } + } + + private static int getInt(Properties properties, String key, int defaultValue) { + String raw = properties.getProperty(key); + if (raw == null || raw.isBlank()) { + return defaultValue; + } + try { + return Integer.parseInt(raw.trim()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Invalid integer value for '" + key + "': " + raw, e); + } + } + + private static double getDouble(Properties properties, String key, double defaultValue) { + String raw = properties.getProperty(key); + if (raw == null || raw.isBlank()) { + return defaultValue; + } + try { + return Double.parseDouble(raw.trim()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Invalid double value for '" + key + "': " + raw, e); + } + } + + private static Duration getDuration(Properties properties, String key, Duration defaultValue) { + String raw = properties.getProperty(key); + if (raw == null || raw.isBlank()) { + return defaultValue; + } + String trimmed = raw.trim(); + // Accept both Flink's human-friendly format (e.g. "1s", "1min") and the ISO-8601 format + // produced by Duration#toString (e.g. "PT1S"). The latter is what HttpDynamicSink emits + // when it serialises a ConfigOption value back into the sink properties bag. + try { + return TimeUtils.parseDuration(trimmed); + } catch (Exception flinkParseError) { + try { + return Duration.parse(trimmed); + } catch (Exception isoParseError) { + IllegalArgumentException ex = + new IllegalArgumentException( + "Invalid duration value for '" + key + "': " + raw, + flinkParseError); + ex.addSuppressed(isoParseError); + throw ex; + } + } + } +} diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java index 0072b709..06a2faa1 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClient.java @@ -27,11 +27,16 @@ import org.apache.flink.connector.http.sink.HttpSinkRequestEntry; import org.apache.flink.connector.http.status.ComposeHttpStatusCodeChecker; import org.apache.flink.connector.http.status.ComposeHttpStatusCodeChecker.ComposeHttpStatusCodeCheckerConfig; +import org.apache.flink.connector.http.status.HttpCodesParser; +import org.apache.flink.connector.http.status.HttpResponseChecker; import org.apache.flink.connector.http.status.HttpStatusCodeChecker; import org.apache.flink.connector.http.utils.HttpHeaderUtils; +import org.apache.flink.util.ConfigurationException; import lombok.extern.slf4j.Slf4j; +import javax.annotation.Nullable; + import java.net.http.HttpClient; import java.util.ArrayList; import java.util.Arrays; @@ -54,6 +59,14 @@ public class JavaNetSinkHttpClient implements SinkHttpClient { private final HttpStatusCodeChecker statusCodeChecker; + /** + * Optional retry-aware response checker. Populated only when the user supplies {@code + * http.sink.retry-codes} or {@code http.sink.success-codes}; when {@code null} the client falls + * back to the legacy {@link #statusCodeChecker} and marks every failure as retryable, + * preserving the historical sink behaviour. + */ + @Nullable private final HttpResponseChecker retryAwareResponseChecker; + private final HttpPostRequestCallback httpPostRequestCallback; private final RequestSubmitter requestSubmitter; @@ -84,6 +97,7 @@ public JavaNetSinkHttpClient( .build(); this.statusCodeChecker = new ComposeHttpStatusCodeChecker(checkerConfig); + this.retryAwareResponseChecker = buildRetryAwareResponseChecker(properties); this.headersAndValues = HttpHeaderUtils.toHeaderAndValueArray(this.headerMap); this.requestSubmitter = @@ -92,6 +106,27 @@ public JavaNetSinkHttpClient( this.httpLogger = HttpLogger.getHttpLogger(properties); } + @Nullable + private static HttpResponseChecker buildRetryAwareResponseChecker(Properties properties) { + String successCodes = + properties.getProperty(HttpConnectorConfigConstants.SINK_RETRY_SUCCESS_CODES); + String retryCodes = + properties.getProperty(HttpConnectorConfigConstants.SINK_RETRY_RETRY_CODES); + if (successCodes == null && retryCodes == null) { + // Neither option was set: keep the legacy behaviour (one-bucket failures) so existing + // users are unaffected. + return null; + } + try { + return new HttpResponseChecker( + HttpCodesParser.parse(successCodes != null ? successCodes : "2XX"), + HttpCodesParser.parse(retryCodes != null ? retryCodes : "500,503,504")); + } catch (ConfigurationException e) { + throw new IllegalArgumentException( + "Invalid HTTP sink success/retry codes configuration: " + e.getMessage(), e); + } + } + @Override public CompletableFuture putRequests( List requestEntries, String endpointUrl) { @@ -114,7 +149,8 @@ private CompletableFuture> submitRequests( private SinkHttpClientResponse prepareSinkHttpClientResponse( List responses, String endpointUrl) { var successfulResponses = new ArrayList(); - var failedResponses = new ArrayList(); + var retryableFailedResponses = new ArrayList(); + var fatalFailedResponses = new ArrayList(); for (var response : responses) { var sinkRequestEntry = response.getHttpRequest(); @@ -123,16 +159,33 @@ private SinkHttpClientResponse prepareSinkHttpClientResponse( httpPostRequestCallback.call( optResponse.orElse(null), sinkRequestEntry, endpointUrl, headerMap); - // TODO Add response processor here and orchestrate it with statusCodeChecker. - if (optResponse.isEmpty() - || statusCodeChecker.isErrorCode(optResponse.get().statusCode())) { - failedResponses.add(sinkRequestEntry); + if (optResponse.isEmpty()) { + // Network-level failure (e.g. IOException) — treat as retryable regardless of the + // active checker so callers can drive their own retry policy. + retryableFailedResponses.add(sinkRequestEntry); + continue; + } + + int statusCode = optResponse.get().statusCode(); + if (retryAwareResponseChecker != null) { + if (retryAwareResponseChecker.isSuccessful(statusCode)) { + successfulResponses.add(sinkRequestEntry); + } else if (retryAwareResponseChecker.isTemporalError(statusCode)) { + retryableFailedResponses.add(sinkRequestEntry); + } else { + fatalFailedResponses.add(sinkRequestEntry); + } + } else if (statusCodeChecker.isErrorCode(statusCode)) { + // Legacy behaviour: everything not marked as success is considered retryable so + // existing users keep the pre-retry-split semantics. + retryableFailedResponses.add(sinkRequestEntry); } else { successfulResponses.add(sinkRequestEntry); } } - return new SinkHttpClientResponse(successfulResponses, failedResponses); + return new SinkHttpClientResponse( + successfulResponses, retryableFailedResponses, fatalFailedResponses); } @VisibleForTesting diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java index c5778bbd..5eee2306 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSink.java @@ -45,8 +45,22 @@ import java.util.Properties; import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MAX_BACKOFF; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MULTIPLIER; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_FIXED_DELAY_DELAY; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_RETRY_CODES; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_STRATEGY_TYPE; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_SUCCESS_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.INSERT_METHOD; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.RETRY_TIMES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_RETRY_CODES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_SUCCESS_CODES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_MULTIPLIER; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_FIXED_DELAY; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_STRATEGY; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.URL; /** @@ -160,6 +174,29 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { .setProperty( SINK_HTTP_RETRY_TIMES, String.valueOf(tableOptions.get(RETRY_TIMES))) + .setProperty( + SINK_RETRY_STRATEGY_TYPE, tableOptions.get(SINK_RETRY_STRATEGY)) + .setProperty( + SINK_RETRY_SUCCESS_CODES, tableOptions.get(SINK_HTTP_SUCCESS_CODES)) + .setProperty( + SINK_RETRY_RETRY_CODES, tableOptions.get(SINK_HTTP_RETRY_CODES)) + .setProperty( + SINK_RETRY_FIXED_DELAY_DELAY, + tableOptions.get(SINK_RETRY_FIXED_DELAY).toString()) + .setProperty( + SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF, + tableOptions + .get(SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF) + .toString()) + .setProperty( + SINK_RETRY_EXP_DELAY_MAX_BACKOFF, + tableOptions + .get(SINK_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF) + .toString()) + .setProperty( + SINK_RETRY_EXP_DELAY_MULTIPLIER, + String.valueOf( + tableOptions.get(SINK_RETRY_EXPONENTIAL_DELAY_MULTIPLIER))) .setProperties(properties); addAsyncOptionsToSinkBuilder(builder); diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java index 54831846..22534417 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicSinkConnectorOptions.java @@ -19,12 +19,20 @@ import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.connector.http.retry.RetryStrategyType; import java.time.Duration; import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES; import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_HTTP_TIMEOUT_SECONDS; import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_REQUEST_CALLBACK_IDENTIFIER; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MAX_BACKOFF; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MULTIPLIER; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_FIXED_DELAY_DELAY; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_RETRY_CODES; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_STRATEGY_TYPE; +import static org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SINK_RETRY_SUCCESS_CODES; /** Table API options for {@link HttpDynamicSink}. */ public class HttpDynamicSinkConnectorOptions { @@ -60,12 +68,64 @@ public class HttpDynamicSinkConnectorOptions { .stringType() .defaultValue(Slf4jHttpPostRequestCallbackFactory.IDENTIFIER); + // ---------- Retry configuration (mirrors the lookup source options) ---------- + public static final ConfigOption RETRY_TIMES = ConfigOptions.key(SINK_HTTP_RETRY_TIMES) .intType() .defaultValue(3) .withDescription( - "Maximum number of retry attempts for HTTP Sink requests on IOException. " - + "Retries use exponential backoff with an initial delay of 1 second. " - + "Set to 0 to disable retries."); + "Maximum number of retry attempts for HTTP Sink requests on " + + "IOException or on a transient HTTP status code (see " + + SINK_RETRY_RETRY_CODES + + "). Set to 0 to disable retries."); + + public static final ConfigOption SINK_RETRY_STRATEGY = + ConfigOptions.key(SINK_RETRY_STRATEGY_TYPE) + .stringType() + .defaultValue(RetryStrategyType.EXPONENTIAL_DELAY.getCode()) + .withDescription( + "Auto retry strategy type for the sink: " + + "fixed-delay or exponential-delay (default)."); + + public static final ConfigOption SINK_HTTP_SUCCESS_CODES = + ConfigOptions.key(SINK_RETRY_SUCCESS_CODES) + .stringType() + .defaultValue("2XX") + .withDescription( + "Comma separated http codes considered as a successful sink response. " + + "Use [1-5]XX for groups and '!' character for excluding."); + + public static final ConfigOption SINK_HTTP_RETRY_CODES = + ConfigOptions.key(SINK_RETRY_RETRY_CODES) + .stringType() + .defaultValue("500,503,504") + .withDescription( + "Comma separated http codes that will trigger a retry when returned by " + + "the sink endpoint. Use [1-5]XX for groups and '!' character " + + "for excluding."); + + public static final ConfigOption SINK_RETRY_FIXED_DELAY = + ConfigOptions.key(SINK_RETRY_FIXED_DELAY_DELAY) + .durationType() + .defaultValue(Duration.ofSeconds(1)) + .withDescription("Fixed-delay interval between sink retries."); + + public static final ConfigOption SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF = + ConfigOptions.key(SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF) + .durationType() + .defaultValue(Duration.ofSeconds(1)) + .withDescription("Exponential-delay initial delay for sink retries."); + + public static final ConfigOption SINK_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF = + ConfigOptions.key(SINK_RETRY_EXP_DELAY_MAX_BACKOFF) + .durationType() + .defaultValue(Duration.ofMinutes(1)) + .withDescription("Exponential-delay maximum delay for sink retries."); + + public static final ConfigOption SINK_RETRY_EXPONENTIAL_DELAY_MULTIPLIER = + ConfigOptions.key(SINK_RETRY_EXP_DELAY_MULTIPLIER) + .doubleType() + .defaultValue(2.0) + .withDescription("Exponential-delay multiplier for sink retries."); } diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java index 14827db6..e44f3758 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java @@ -34,7 +34,14 @@ import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.INSERT_METHOD; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.REQUEST_CALLBACK_IDENTIFIER; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.RETRY_TIMES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_RETRY_CODES; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_HTTP_SUCCESS_CODES; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_EXPONENTIAL_DELAY_MULTIPLIER; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_FIXED_DELAY; +import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.SINK_RETRY_STRATEGY; import static org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnectorOptions.URL; /** Factory for creating {@link HttpDynamicSink}. */ @@ -100,6 +107,13 @@ public Set> optionalOptions() { options.add(SINK_REQUEST_TIMEOUT); options.add(REQUEST_CALLBACK_IDENTIFIER); options.add(RETRY_TIMES); + options.add(SINK_RETRY_STRATEGY); + options.add(SINK_HTTP_SUCCESS_CODES); + options.add(SINK_HTTP_RETRY_CODES); + options.add(SINK_RETRY_FIXED_DELAY); + options.add(SINK_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF); + options.add(SINK_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF); + options.add(SINK_RETRY_EXPONENTIAL_DELAY_MULTIPLIER); return options; } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java index 6db7838a..7f9c1fa7 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java @@ -257,4 +257,58 @@ public void testNoRetryWhenDisabled() throws InterruptedException { // requestResult.accept() called exactly once assertThat(acceptCallCount.get()).isEqualTo(1); } + + @Test + public void testFatalFailuresAreNotRetried() throws InterruptedException { + // Fatal failures (non-retryable status code, e.g. 404) must be counted immediately + // without triggering any retry. + HttpSinkRequestEntry request = new HttpSinkRequestEntry("PUT", "hello".getBytes()); + List requestEntries = Collections.singletonList(request); + + HttpRequest mockHttpRequest = + new HttpRequest(null, Collections.singletonList("hello".getBytes()), "PUT"); + + SinkHttpClientResponse fatalResponse = + new SinkHttpClientResponse( + Collections.emptyList(), + Collections.emptyList(), + Collections.singletonList(mockHttpRequest)); + + when(httpClient.putRequests(anyList(), anyString())) + .thenReturn(CompletableFuture.completedFuture(fatalResponse)); + + Properties properties = new Properties(); + properties.setProperty( + org.apache.flink.connector.http.config.HttpConnectorConfigConstants + .SINK_HTTP_RETRY_TIMES, + "3"); + + Collection> stateBuffer = new ArrayList<>(); + HttpSinkWriter writer = + new HttpSinkWriter<>( + elementConverter, + context, + 10, + 10, + 100, + 10, + 10, + 10, + "http://localhost/client", + httpClient, + stateBuffer, + properties); + + AtomicInteger acceptCallCount = new AtomicInteger(0); + Consumer> requestResult = + httpSinkRequestEntries -> acceptCallCount.incrementAndGet(); + + writer.submitRequestEntries(requestEntries, requestResult); + Thread.sleep(1500); + + // Only 1 attempt — fatal failures are not retried even though retry.times=3. + verify(httpClient, times(1)).putRequests(anyList(), anyString()); + verify(errorCounter).inc(requestEntries.size()); + assertThat(acceptCallCount.get()).isEqualTo(1); + } } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/SinkRetryConfigTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/SinkRetryConfigTest.java new file mode 100644 index 00000000..0a8c5215 --- /dev/null +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/SinkRetryConfigTest.java @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.http.sink; + +import org.apache.flink.connector.http.config.HttpConnectorConfigConstants; + +import org.junit.jupiter.api.Test; + +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Unit tests for {@link SinkRetryConfig}. */ +class SinkRetryConfigTest { + + @Test + void shouldUseDefaultsWhenNoPropertiesSet() { + var cfg = SinkRetryConfig.fromProperties(new Properties()); + + assertThat(cfg.getMaxRetries()).isEqualTo(SinkRetryConfig.DEFAULT_MAX_RETRIES); + // Default success checker treats 2xx as success and {500,503,504} as retryable. + assertThat(cfg.isSuccess(200)).isTrue(); + assertThat(cfg.isSuccess(201)).isTrue(); + assertThat(cfg.shouldRetry(500)).isTrue(); + assertThat(cfg.shouldRetry(503)).isTrue(); + assertThat(cfg.shouldRetry(504)).isTrue(); + // 404 is neither success nor retryable — fatal. + assertThat(cfg.isSuccess(404)).isFalse(); + assertThat(cfg.shouldRetry(404)).isFalse(); + } + + @Test + void shouldHonourFixedDelayStrategy() { + Properties props = new Properties(); + props.setProperty(HttpConnectorConfigConstants.SINK_HTTP_RETRY_TIMES, "5"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_STRATEGY_TYPE, "fixed-delay"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_FIXED_DELAY_DELAY, "250ms"); + + var cfg = SinkRetryConfig.fromProperties(props); + + assertThat(cfg.getMaxRetries()).isEqualTo(5); + // Fixed delay — every attempt returns the same value. + assertThat(cfg.backoffMillis(0)).isEqualTo(250L); + assertThat(cfg.backoffMillis(3)).isEqualTo(250L); + } + + @Test + void shouldHonourExponentialDelayStrategy() { + Properties props = new Properties(); + props.setProperty( + HttpConnectorConfigConstants.SINK_RETRY_STRATEGY_TYPE, "exponential-delay"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF, "1s"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MAX_BACKOFF, "10s"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MULTIPLIER, "2.0"); + + var cfg = SinkRetryConfig.fromProperties(props); + + assertThat(cfg.backoffMillis(0)).isEqualTo(1000L); + assertThat(cfg.backoffMillis(1)).isEqualTo(2000L); + assertThat(cfg.backoffMillis(2)).isEqualTo(4000L); + // Eventually capped at the configured max (10s). + assertThat(cfg.backoffMillis(10)).isEqualTo(10_000L); + } + + @Test + void shouldAcceptIsoDurationsAlongsideFlinkFormat() { + // HttpDynamicSink writes ConfigOption values as Duration#toString (ISO-8601), + // so the parser must accept both formats. + Properties props = new Properties(); + props.setProperty( + HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_INITIAL_BACKOFF, "PT2S"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_EXP_DELAY_MULTIPLIER, "3.0"); + + var cfg = SinkRetryConfig.fromProperties(props); + + assertThat(cfg.backoffMillis(0)).isEqualTo(2000L); + } + + @Test + void shouldHonourCustomSuccessAndRetryCodes() { + Properties props = new Properties(); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_SUCCESS_CODES, "2XX,201"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_RETRY_CODES, "429,502,503"); + + var cfg = SinkRetryConfig.fromProperties(props); + + assertThat(cfg.shouldRetry(429)).isTrue(); + assertThat(cfg.shouldRetry(502)).isTrue(); + assertThat(cfg.shouldRetry(500)).isFalse(); + assertThat(cfg.isSuccess(200)).isTrue(); + } + + @Test + void shouldRejectInvalidStrategy() { + Properties props = new Properties(); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_STRATEGY_TYPE, "not-a-strategy"); + + assertThatThrownBy(() -> SinkRetryConfig.fromProperties(props)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void shouldRejectInvalidDuration() { + Properties props = new Properties(); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_FIXED_DELAY_DELAY, "???"); + props.setProperty(HttpConnectorConfigConstants.SINK_RETRY_STRATEGY_TYPE, "fixed-delay"); + + assertThatThrownBy(() -> SinkRetryConfig.fromProperties(props)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(HttpConnectorConfigConstants.SINK_RETRY_FIXED_DELAY_DELAY); + } +}