diff --git a/data-prepper-plugins/sqs-common/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoff.java b/data-prepper-plugins/sqs-common/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoff.java index b5f85cd61d..3e9ae82376 100644 --- a/data-prepper-plugins/sqs-common/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoff.java +++ b/data-prepper-plugins/sqs-common/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoff.java @@ -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); } diff --git a/data-prepper-plugins/sqs-common/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoffTest.java b/data-prepper-plugins/sqs-common/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoffTest.java index b793cf39bc..872e96eb24 100644 --- a/data-prepper-plugins/sqs-common/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoffTest.java +++ b/data-prepper-plugins/sqs-common/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/common/SqsBackoffTest.java @@ -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 invalidMaximumDelays() { + return Stream.of( + Arguments.of((Duration) null), + Arguments.of(Duration.ZERO), + Arguments.of(Duration.ofSeconds(-1)), + Arguments.of(Duration.ofSeconds(5)) ); } } diff --git a/data-prepper-plugins/sqs-source/README.md b/data-prepper-plugins/sqs-source/README.md index c3898b396d..7115a898c1 100644 --- a/data-prepper-plugins/sqs-source/README.md +++ b/data-prepper-plugins/sqs-source/README.md @@ -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. diff --git a/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfig.java b/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfig.java index 6d00c68d06..c106507dbc 100644 --- a/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfig.java +++ b/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfig.java @@ -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; + } } diff --git a/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsService.java b/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsService.java index 3cf6b3942b..aefa576458 100644 --- a/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsService.java +++ b/data-prepper-plugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsService.java @@ -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; diff --git a/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfigTest.java b/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfigTest.java index c6bdb8b644..1864a24368 100644 --- a/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfigTest.java +++ b/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfigTest.java @@ -10,15 +10,33 @@ 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.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasItem; +import static org.hamcrest.Matchers.hasProperty; +import static org.hamcrest.Matchers.hasToString; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; 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 +51,33 @@ 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"); + assertThat(queueConfig.getMaximumBackoff(), equalTo(Duration.ofMinutes(5))); + } + + @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); + + assertThat(queueConfig.getMaximumBackoff(), equalTo(Duration.ofMinutes(10))); + } + + @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> violations = validator.validate(queueConfig); + assertThat(violations, hasItem(hasProperty("propertyPath", hasToString("maximumBackoff")))); + } } -} \ No newline at end of file +} diff --git a/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/SqsServiceTest.java b/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/SqsServiceTest.java index c317852903..8855fb132c 100644 --- a/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/SqsServiceTest.java +++ b/data-prepper-plugins/sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/SqsServiceTest.java @@ -9,22 +9,28 @@ package org.opensearch.dataprepper.plugins.source.sqs; +import com.linecorp.armeria.client.retry.Backoff; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; import org.opensearch.dataprepper.metrics.PluginMetrics; import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager; import org.opensearch.dataprepper.model.buffer.Buffer; import org.opensearch.dataprepper.model.event.Event; import org.opensearch.dataprepper.model.record.Record; +import org.opensearch.dataprepper.plugins.source.sqs.common.SqsBackoff; 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.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; - class SqsServiceTest { private SqsSourceConfig sqsSourceConfig; private PluginMetrics pluginMetrics; @@ -46,23 +52,43 @@ void setUp() { @Test void start_with_single_queue_starts_workers() { + final Duration configuredBackoff = Duration.ofMinutes(10); 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(configuredBackoff); when(sqsSourceConfig.getQueues()).thenReturn(List.of(queueConfig)); - SqsService sqsService = spy(new SqsService(buffer, acknowledgementSetManager, sqsSourceConfig, pluginMetrics, credentialsProvider)); - sqsService.start(); + + try (final MockedStatic sqsBackoffMockedStatic = mockStatic(SqsBackoff.class)) { + sqsBackoffMockedStatic.when(() -> SqsBackoff.createExponentialBackoff(any(Duration.class))) + .thenReturn(mock(Backoff.class)); + + SqsService sqsService = spy(new SqsService(buffer, acknowledgementSetManager, sqsSourceConfig, pluginMetrics, credentialsProvider)); + sqsService.start(); + + sqsBackoffMockedStatic.verify(() -> SqsBackoff.createExponentialBackoff(configuredBackoff)); + } } @Test void stop_should_shutdown_executors_and_workers() throws InterruptedException { + final Duration configuredBackoff = Duration.ofMinutes(3); 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(configuredBackoff); when(sqsSourceConfig.getQueues()).thenReturn(List.of(queueConfig)); - SqsService sqsService = new SqsService(buffer, acknowledgementSetManager, sqsSourceConfig, pluginMetrics, credentialsProvider) {}; - sqsService.start(); - sqsService.stop(); + + try (final MockedStatic sqsBackoffMockedStatic = mockStatic(SqsBackoff.class)) { + sqsBackoffMockedStatic.when(() -> SqsBackoff.createExponentialBackoff(any(Duration.class))) + .thenReturn(mock(Backoff.class)); + + SqsService sqsService = new SqsService(buffer, acknowledgementSetManager, sqsSourceConfig, pluginMetrics, credentialsProvider); + sqsService.start(); + sqsService.stop(); + + sqsBackoffMockedStatic.verify(() -> SqsBackoff.createExponentialBackoff(configuredBackoff)); + } } } \ No newline at end of file