[FLINK-33702][datastream] Add IncrementalDelayRetryStrategy in AsyncR…

…etryStrategies

flink-streaming-java/src/main/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategies.java

@@ -237,4 +237,91 @@ public ExponentialBackoffDelayRetryStrategy<OUT> build() {
                     exceptionPredicate);
         }
     }
+
+    /** IncrementalDelayRetryStrategy. */
+    public static class IncrementalDelayRetryStrategy<OUT> implements AsyncRetryStrategy<OUT> {
+        private static final long serialVersionUID = 1L;
+        private final int maxAttempts;
+        private long initialDelay;
+        private final long maxRetryDelay;
+        private final long incremental;
+        private final Predicate<Collection<OUT>> resultPredicate;
+        private final Predicate<Throwable> exceptionPredicate;
+
+        private IncrementalDelayRetryStrategy(
+                int maxAttempts,
+                long initialDelay,
+                long maxRetryDelay,
+                long incremental,
+                Predicate<Collection<OUT>> resultPredicate,
+                Predicate<Throwable> exceptionPredicate) {
+            this.maxAttempts = maxAttempts;
+            this.initialDelay = initialDelay;
+            this.maxRetryDelay = maxRetryDelay;
+            this.incremental = incremental;
+            this.resultPredicate = resultPredicate;
+            this.exceptionPredicate = exceptionPredicate;
+        }
+
+        @Override
+        public boolean canRetry(int currentAttempts) {
+            return currentAttempts <= maxAttempts;
+        }
+
+        @Override
+        public AsyncRetryPredicate<OUT> getRetryPredicate() {
+            return new RetryPredicate(resultPredicate, exceptionPredicate);
+        }
+
+        @Override
+        public long getBackoffTimeMillis(int currentAttempts) {
+            if (currentAttempts <= 1) {
+                // equivalent to initial delay
+                return initialDelay;
+            }
+
+            return Math.min(initialDelay + (currentAttempts - 1) * incremental, maxRetryDelay);
+        }
+    }
+
+    /** IncrementalDelayRetryStrategyBuilder for building a IncrementalDelayRetryStrategy. */
+    public static class IncrementalDelayRetryStrategyBuilder<OUT> {
+        private final int maxAttempts;
+        private final long initialDelay;
+        private final long maxRetryDelay;
+        private final long incremental;
+
+        private Predicate<Collection<OUT>> resultPredicate;
+        private Predicate<Throwable> exceptionPredicate;
+
+        public IncrementalDelayRetryStrategyBuilder(
+                int maxAttempts, long initialDelay, long maxRetryDelay, long incremental) {
+            this.maxAttempts = maxAttempts;
+            this.initialDelay = initialDelay;
+            this.maxRetryDelay = maxRetryDelay;
+            this.incremental = incremental;
+        }
+
+        public IncrementalDelayRetryStrategyBuilder<OUT> ifResult(
+                @Nonnull Predicate<Collection<OUT>> resultRetryPredicate) {
+            this.resultPredicate = resultRetryPredicate;
+            return this;
+        }
+
+        public IncrementalDelayRetryStrategyBuilder<OUT> ifException(
+                @Nonnull Predicate<Throwable> exceptionRetryPredicate) {
+            this.exceptionPredicate = exceptionRetryPredicate;
+            return this;
+        }
+
+        public IncrementalDelayRetryStrategy<OUT> build() {
+            return new IncrementalDelayRetryStrategy<OUT>(
+                    maxAttempts,
+                    initialDelay,
+                    maxRetryDelay,
+                    incremental,
+                    resultPredicate,
+                    exceptionPredicate);
+        }
+    }
 }

flink-streaming-java/src/test/java/org/apache/flink/streaming/util/retryable/AsyncRetryStrategiesTest.java

@@ -0,0 +1,42 @@
+/*
+ * 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.streaming.util.retryable;
+
+import org.apache.flink.streaming.api.functions.async.AsyncRetryStrategy;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/** Tests for the {@link AsyncRetryStrategies}. */
+public class AsyncRetryStrategiesTest extends TestLogger {
+
+    @Test
+    public void testIncrementalDelayRetryStrategy() {
+        AsyncRetryStrategy<Void> IncrementalDelayRetryStrategy =
+                new AsyncRetryStrategies.IncrementalDelayRetryStrategyBuilder<Void>(
+                                10, 100, 500, 100)
+                        .build();
+
+        Assert.assertEquals(100, IncrementalDelayRetryStrategy.getBackoffTimeMillis(0));
+        Assert.assertEquals(100, IncrementalDelayRetryStrategy.getBackoffTimeMillis(1));
+        Assert.assertEquals(500, IncrementalDelayRetryStrategy.getBackoffTimeMillis(5));
+        Assert.assertEquals(500, IncrementalDelayRetryStrategy.getBackoffTimeMillis(6));
+    }
+}