[FLINK-39366][connector-http] Add retry support with exponential backoff for HTTP Sink

docs/content.zh/docs/connectors/table/http.md

@@ -547,6 +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) 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.

docs/content/docs/connectors/table/http.md

@@ -547,6 +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) 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.

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.
+ *
+ * <p>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<HttpRequest> successfulRequests;
 
-    /** A list of requests that {@link SinkHttpClient} failed to write. */
-    @NonNull private final List<HttpRequest> failedRequests;
+    /** Requests that failed with a transient HTTP status code and may be retried. */
+    @NonNull private final List<HttpRequest> retryableFailedRequests;
+
+    /** Requests that failed with a non-retryable status code (fatal failures). */
+    @NonNull private final List<HttpRequest> fatalFailedRequests;
+
+    public SinkHttpClientResponse(
+            @NonNull List<HttpRequest> successfulRequests,
+            @NonNull List<HttpRequest> retryableFailedRequests,
+            @NonNull List<HttpRequest> 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<HttpRequest> successfulRequests,
+            @NonNull List<HttpRequest> 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<HttpRequest> getFailedRequests() {
+        List<HttpRequest> all =
+                new ArrayList<>(retryableFailedRequests.size() + fatalFailedRequests.size());
+        all.addAll(retryableFailedRequests);
+        all.addAll(fatalFailedRequests);
+        return all;
+    }
 }

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 ------
@@ -164,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";
 }

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,52 +25,125 @@
 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;
 import static org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MULTIPLIER;
 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.
+ *
+ * <p>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<String> strategy;
+        private final ConfigOption<Duration> fixedDelay;
+        private final ConfigOption<Duration> exponentialInitialBackoff;
+        private final ConfigOption<Duration> exponentialMaxBackoff;
+        private final ConfigOption<Double> exponentialMultiplier;
+
+        public RetryOptionKeys(
+                ConfigOption<String> strategy,
+                ConfigOption<Duration> fixedDelay,
+                ConfigOption<Duration> exponentialInitialBackoff,
+                ConfigOption<Duration> exponentialMaxBackoff,
+                ConfigOption<Double> 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<String> strategy() {
+            return strategy;
+        }
+
+        ConfigOption<Duration> fixedDelay() {
+            return fixedDelay;
+        }
+
+        ConfigOption<Duration> 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<Duration> exponentialMaxBackoff() {
+            return exponentialMaxBackoff;
+        }
+
+        ConfigOption<Double> exponentialMultiplier() {
+            return exponentialMultiplier;
+        }
     }
 }