LuisNorman · Oct 4, 2026 · Oct 13, 2026 · Oct 29, 2026 · srikanthpadakanti · Oct 12, 2026
@@ -13,14 +13,38 @@
 import java.time.Duration;
 
 public final class SqsBackoff {
-    private static final long INITIAL_DELAY_MILLIS = Duration.ofSeconds(20).toMillis();
-    private static final long MAX_DELAY_MILLIS = Duration.ofMinutes(5).toMillis();
-    private static final double JITTER_RATE = 0.20;
+    static final long INITIAL_DELAY_MILLIS = Duration.ofSeconds(20).toMillis();
+    static final long DEFAULT_MAX_DELAY_MILLIS = Duration.ofMinutes(5).toMillis();
+    static final double JITTER_RATE = 0.20;
 
     private SqsBackoff() {}
 
+    /**
+     * Creates an exponential backoff that uses the default maximum delay.
+     *
+     * @return a configured {@link Backoff}
+     */
     public static Backoff createExponentialBackoff() {
-        return Backoff.exponential(INITIAL_DELAY_MILLIS, MAX_DELAY_MILLIS)
+        return createExponentialBackoff(Duration.ofMillis(DEFAULT_MAX_DELAY_MILLIS));
+    }
+
+    /**
+     * Creates an exponential backoff capped at {@code maximumDelay}.
+     *
+     * @param maximumDelay the maximum delay between retries. Must be non-null and at
+     *                     least the backoff's initial delay of 20 seconds.
+     * @return a configured {@link Backoff}
+     * @throws IllegalArgumentException if {@code maximumDelay} is null or shorter than
+     *                                  the initial delay
+     */
+    public static Backoff createExponentialBackoff(final Duration maximumDelay) {
+        // Armeria rejects a maximum delay smaller than the initial delay.
+        if (maximumDelay == null || maximumDelay.toMillis() < INITIAL_DELAY_MILLIS) {
+            throw new IllegalArgumentException(
+                    "maximumDelay must be at least " + Duration.ofMillis(INITIAL_DELAY_MILLIS)
+                            + ", but was " + maximumDelay);
+        }
+        return Backoff.exponential(INITIAL_DELAY_MILLIS, maximumDelay.toMillis())
                 .withJitter(JITTER_RATE)
                 .withMaxAttempts(Integer.MAX_VALUE);
     }

@@ -11,25 +11,64 @@
 
 import com.linecorp.armeria.client.retry.Backoff;
 import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.time.Duration;
+import java.util.stream.Stream;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 class SqsBackoffTest {
 
     @Test
     void testCreateExponentialBackoff() {
         final Backoff backoff = SqsBackoff.createExponentialBackoff();
-        assertNotNull(backoff, "Backoff should not be null");
+        assertThat("Backoff should not be null", backoff, is(notNullValue()));
+
         final long firstDelay = backoff.nextDelayMillis(1);
-        final long expectedBaseDelay = 20_000L;
-        final double jitterRate = 0.20;
-        final long minDelay = (long) (expectedBaseDelay * (1 - jitterRate));
-        final long maxDelay = (long) (expectedBaseDelay * (1 + jitterRate));
-
-        assertTrue(
-                firstDelay >= minDelay && firstDelay <= maxDelay,
-                String.format("First delay %dms should be between %dms and %dms",
-                        firstDelay, minDelay, maxDelay)
+        final long minDelay = (long) (SqsBackoff.INITIAL_DELAY_MILLIS * (1 - SqsBackoff.JITTER_RATE));
+        final long maxDelay = (long) (SqsBackoff.INITIAL_DELAY_MILLIS * (1 + SqsBackoff.JITTER_RATE));
+        assertThat("First delay should be at or above the default initial delay lower bound",
+                firstDelay, is(greaterThanOrEqualTo(minDelay)));
+        assertThat("First delay should be at or below the default initial delay upper bound",
+                firstDelay, is(lessThanOrEqualTo(maxDelay)));
+    }
+
+    @Test
+    void testCreateExponentialBackoff_capsAtConfiguredMaximumDelay() {
+        final Duration maximumDelay = Duration.ofMinutes(1);
+        final Backoff backoff = SqsBackoff.createExponentialBackoff(maximumDelay);
+        assertThat("Backoff should not be null", backoff, is(notNullValue()));
+
+        // After many attempts the exponential delay should be clamped to the configured maximum.
+        final long maxDelayCeiling = (long) (maximumDelay.toMillis() * (1 + SqsBackoff.JITTER_RATE));
+        final long delayAtHighAttempt = backoff.nextDelayMillis(50);
+        assertThat("Delay at high attempt should be at or below the configured maximum ceiling",
+                delayAtHighAttempt, is(lessThanOrEqualTo(maxDelayCeiling)));
+    }
+
+    @ParameterizedTest(name = "throws when maximum delay is {0}")
+    @MethodSource("invalidMaximumDelays")
+    void testCreateExponentialBackoff_rejectsInvalidMaximumDelay(final Duration invalidMaximumDelay) {
+        // Armeria rejects a maximum delay smaller than the initial delay, so the factory must
+        // fail fast rather than silently substituting a default and masking a programmer bug.
+        assertThrows(IllegalArgumentException.class,
+                () -> SqsBackoff.createExponentialBackoff(invalidMaximumDelay));
+    }
+
+    private static Stream<Arguments> invalidMaximumDelays() {
+        return Stream.of(
+                Arguments.of((Duration) null),
+                Arguments.of(Duration.ZERO),
+                Arguments.of(Duration.ofSeconds(-1)),
+                Arguments.of(Duration.ofSeconds(5))
         );
     }
 }
@@ -30,6 +30,7 @@ sqs-pipeline:
           visibility_duplication_protection: true
           visibility_duplicate_protection_timeout: "PT1H"
           on_error: "retain_messages"
+          maximum_backoff: "PT5M"
           codec:
             json:
               key_name: "events"
@@ -54,6 +55,10 @@ sqs-pipeline:
   Use an `on_error` option to control behavior on errors (e.g., delete or retain messages)
 
 
+- **Configurable Backoff:**  
+  Tune the exponential retry delay ceiling with `maximum_backoff`. Default is 5 minutes.
+
+
 - **Codec Support:**  
   Configure codecs (e.g., JSON, CSV, newline-delimited) to parse incoming messages.
 

@@ -30,6 +30,7 @@ public class QueueConfig {
     private static final Duration DEFAULT_VISIBILITY_DUPLICATE_PROTECTION_TIMEOUT = Duration.ofHours(2);
     private static final Duration DEFAULT_WAIT_TIME_SECONDS = null;
     private static final Duration DEFAULT_POLL_DELAY_SECONDS = Duration.ofSeconds(0);
+    private static final Duration DEFAULT_MAXIMUM_BACKOFF = Duration.ofMinutes(5);
     static final int DEFAULT_NUMBER_OF_WORKERS = 1;
 
     @JsonProperty("url")
@@ -75,6 +76,11 @@ public class QueueConfig {
     @JsonProperty("on_error")
     private OnErrorOption onErrorOption = OnErrorOption.RETAIN_MESSAGES;
 
+    // Must be at least SqsBackoff's initial delay (20s) or Armeria rejects the backoff at startup.
+    @JsonProperty("maximum_backoff")
+    @DurationMin(seconds = 20)
+    private Duration maximumBackoff = DEFAULT_MAXIMUM_BACKOFF;
+
     public String getUrl() {
         return url;
     }
@@ -112,5 +118,9 @@ public InputCodec getCodec() {
     public OnErrorOption getOnErrorOption() {
         return onErrorOption;
     }
+
+    public Duration getMaximumBackoff() {
+        return maximumBackoff;
+    }
 }
 
@@ -71,7 +71,7 @@ public void start() {
             SqsClient sqsClient = sqsClientMap.computeIfAbsent(region,
                     r -> SqsClientFactory.createSqsClient(Region.of(r), credentialsProvider));
             String queueName = queueUrl.substring(queueUrl.lastIndexOf('/') + 1);
-            Backoff backoff = SqsBackoff.createExponentialBackoff();
+            Backoff backoff = SqsBackoff.createExponentialBackoff(queueConfig.getMaximumBackoff());
             SqsWorkerCommon sqsWorkerCommon = new SqsWorkerCommon(backoff, pluginMetrics, acknowledgementSetManager);
             int numWorkers = queueConfig.getNumWorkers();
             SqsEventProcessor sqsEventProcessor;

@@ -10,15 +10,29 @@
 
 package org.opensearch.dataprepper.plugins.source.sqs;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import jakarta.validation.ConstraintViolation;
+import jakarta.validation.Validation;
+import jakarta.validation.Validator;
+import jakarta.validation.ValidatorFactory;
+import org.hibernate.validator.messageinterpolation.ParameterMessageInterpolator;
 import org.junit.jupiter.api.Test;
+
 import java.time.Duration;
+import java.util.Set;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class QueueConfigTest {
 
+    private static final ObjectMapper YAML_MAPPER =
+            new ObjectMapper(new YAMLFactory()).registerModule(new JavaTimeModule());
+
     @Test
     void testDefaultValues() {
         final QueueConfig queueConfig = new QueueConfig();
@@ -33,5 +47,37 @@ void testDefaultValues() {
         assertEquals(Duration.ofHours(2), queueConfig.getVisibilityDuplicateProtectionTimeout(),
                 "Visibility duplicate protection timeout should default to 2 hours");
         assertNull(queueConfig.getWaitTime(), "Wait time should default to null");
+        assertEquals(Duration.ofMinutes(5), queueConfig.getMaximumBackoff(),
+                "Maximum backoff should default to 5 minutes");
+    }
+
+    @Test
+    void testMaximumBackoffReadsConfiguredValue() throws Exception {
+        final String yaml = "url: \"https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue\"\n"
+                + "maximum_backoff: \"PT10M\"\n";
+
+        final QueueConfig queueConfig = YAML_MAPPER.readValue(yaml, QueueConfig.class);
+
+        assertEquals(Duration.ofMinutes(10), queueConfig.getMaximumBackoff(),
+                "Maximum backoff should reflect the configured value");
+    }
+
+    @Test
+    void testMaximumBackoffBelowMinimumFailsValidation() throws Exception {
+        final String yaml = "url: \"https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue\"\n"
+                + "maximum_backoff: \"PT10S\"\n";
+
+        final QueueConfig queueConfig = YAML_MAPPER.readValue(yaml, QueueConfig.class);
+
+        // ParameterMessageInterpolator avoids requiring a Jakarta EL provider on the test classpath.
+        try (final ValidatorFactory factory = Validation.byDefaultProvider().configure()
+                .messageInterpolator(new ParameterMessageInterpolator())
+                .buildValidatorFactory()) {
+            final Validator validator = factory.getValidator();
+            final Set<ConstraintViolation<QueueConfig>> violations = validator.validate(queueConfig);
+            assertTrue(
+                    violations.stream().anyMatch(v -> v.getPropertyPath().toString().equals("maximumBackoff")),
+                    "Expected a validation violation on maximumBackoff for a value below 20 seconds");
+        }
     }
-}
+}
@@ -19,6 +19,7 @@
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.regions.Region;
 
+import java.time.Duration;
 import java.util.List;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -49,6 +50,7 @@ void start_with_single_queue_starts_workers() {
         QueueConfig queueConfig = mock(QueueConfig.class);
         when(queueConfig.getUrl()).thenReturn("https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue");
         when(queueConfig.getNumWorkers()).thenReturn(2);
+        when(queueConfig.getMaximumBackoff()).thenReturn(Duration.ofMinutes(5));
         when(sqsSourceConfig.getQueues()).thenReturn(List.of(queueConfig));
         SqsService sqsService = spy(new SqsService(buffer, acknowledgementSetManager, sqsSourceConfig, pluginMetrics, credentialsProvider));
         sqsService.start();
@@ -59,6 +61,7 @@ void stop_should_shutdown_executors_and_workers() throws InterruptedException {
         QueueConfig queueConfig = mock(QueueConfig.class);
         when(queueConfig.getUrl()).thenReturn("https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue");
         when(queueConfig.getNumWorkers()).thenReturn(1);
+        when(queueConfig.getMaximumBackoff()).thenReturn(Duration.ofMinutes(5));
         when(sqsSourceConfig.getQueues()).thenReturn(List.of(queueConfig));
         SqsService sqsService = new SqsService(buffer, acknowledgementSetManager, sqsSourceConfig, pluginMetrics, credentialsProvider) {};
         sqsService.start();