Skip to content

feat(sqs-source): make maximum backoff configurable per queue - #7186

Open
LuisNorman wants to merge 3 commits into
opensearch-project:mainfrom
LuisNorman:sqs-maximum-backoff
Open

LuisNorman wants to merge 3 commits into
opensearch-project:mainfrom
LuisNorman:sqs-maximum-backoff

Conversation

@LuisNorman

@LuisNorman LuisNorman commented Oct 7, 2026 •

Copy link
Copy Markdown
Contributor

Description

Adds a new per-queue maximum_backoff option on the sqs source that caps the exponential retry delay used when the source cannot read from Amazon SQS. The default stays at 5 minutes so existing deployments see no behavior change.

Operators of pipelines pulling from Amazon SQS sometimes need to shorten or lengthen the retry ceiling. Lower it to recover faster from transient failures. Raise it to be gentler during longer outages. The option exposes that lever without changing the shared SqsBackoff factory's default.

Changes:

  • New maximum_backoff option on QueueConfig with @DurationMin(seconds = 20). The floor matches SqsBackoff.INITIAL_DELAY_MILLIS. Armeria's Backoff.exponential throws if max is below initial.
  • New SqsBackoff.createExponentialBackoff(Duration) overload that throws IllegalArgumentException on null or sub-initial inputs so misuse surfaces at startup instead of being masked by the default.
  • SqsService constructs the backoff per queue from queueConfig.getMaximumBackoff().
  • Unit tests cover the baseline, configured-ceiling, and four invalid-input rejection cases (parameterized).

Issues Resolved

Resolves #3744

Follow-up: a companion PR will extend the same maximum_backoff option to the s3 source, which shares the underlying SqsBackoff factory.

Check List

  • New functionality includes testing.
  • New functionality has a documentation issue. Please link to it in this PR.
    • New functionality has javadoc added
  • Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Adds a new per-queue maximum_backoff option on the SQS source that
caps the exponential retry delay used when the source cannot read
from SQS. The default stays at 5 minutes so existing deployments
see no behavior change.

Also makes SqsBackoff resilient to null, zero, and negative durations
by falling back to the default, so the shared factory stays safe for
all callers of the parameterless overload.

Signed-off-by: Luis Norman <lunorman@amazon.com>
@github-actions

github-actions Bot commented Oct 7, 2026 •

Copy link
Copy Markdown

PR Reviewer Guide 🔍

(Review updated until commit 4abb732)

Here are some key observations to aid the review process:

🧪 PR contains tests
🔒 No security concerns identified
✅ No TODO sections
🔀 No multiple PR themes
⚡ No major issues detected

* @return a configured {@link Backoff}
*/
public static Backoff createExponentialBackoff(final Duration maximumDelay) {
final long maxDelayMillis = isValidMaximumDelay(maximumDelay) ? maximumDelay.toMillis() : DEFAULT_MAX_DELAY_MILLIS;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unreachable from config given @DurationMin and silent fallback hides misconfiguration. Throw or log instead?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. I'll update it to throw IllegalArgumentException on an invalid input.

r -> SqsClientFactory.createSqsClient(Region.of(r), credentialsProvider));
String queueName = queueUrl.substring(queueUrl.lastIndexOf('/') + 1);
Backoff backoff = SqsBackoff.createExponentialBackoff();
Backoff backoff = SqsBackoff.createExponentialBackoff(queueConfig.getMaximumBackoff());

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No test that maximum_backoff reaches the backoff, or that values under 20s are rejected.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for catching this. I'll add tests to cover both.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for adding the tests, @LuisNorman! One small gap: SqsServiceTest doesn't use MockitoExtension, so the getMaximumBackoff() stub is lenient, and the tests still pass if this line goes back to createExponentialBackoff(). Could you add a check that the configured maximum_backoff actually reaches SqsBackoff? A MockedStatic on SqsBackoff with a verify on the configured Duration would do it.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. Will wrap the test in MockedStatic<SqsBackoff> and verify the configured Duration reaches the factory.

@srikanthpadakanti srikanthpadakanti left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hello @LuisNorman Thank you for this contribution. Can you please address my comments.

Addresses review feedback on PR opensearch-project#7186.

- SqsBackoff.createExponentialBackoff(Duration) now throws
  IllegalArgumentException on null or sub-initial values instead of
  silently falling back to the default. @DurationMin on QueueConfig
  already blocks invalid user input, so anything reaching the factory
  with a bad value is a programming bug.
- Adds QueueConfigTest cases for a configured maximum_backoff reaching
  getMaximumBackoff() and for bean validation rejecting sub-20s values.
- Stubs getMaximumBackoff() on SqsServiceTest's QueueConfig mocks to
  satisfy the new precondition.

Signed-off-by: Luis Norman <lunorman@amazon.com>
@github-actions

Copy link
Copy Markdown

Persistent review updated to latest commit 56040d1

@LuisNorman

Copy link
Copy Markdown
Contributor Author

Hi @srikanthpadakanti, the follow-up commit addresses both review comments and CI is green. Happy to make any other changes you'd like.

.buildValidatorFactory()) {
final Validator validator = factory.getValidator();
final Set<ConstraintViolation<QueueConfig>> violations = validator.validate(queueConfig);
assertTrue(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Small style request: Data Prepper prefers Hamcrest for new assertions. Could you switch the new ones here (lines 50, 61 and 78) to assertThat(...)? For example, assertThat(queueConfig.getMaximumBackoff(), equalTo(Duration.ofMinutes(10))), and for the violations, assertThat(violations, hasItem(...)) on the property path. That would match what you already did nicely in SqsBackoffTest. Thanks!

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will switch the three new assertions to Hamcrest.

@srikanthpadakanti

Copy link
Copy Markdown
Collaborator

Thanks for the quick turnaround on the earlier feedback! A minor note: the description still says SqsBackoff has a "defensive fallback" for invalid inputs, but it now throws IllegalArgumentException. Could you update the description so it matches the code?

@LuisNorman

Copy link
Copy Markdown
Contributor Author

Will update the description to match.

… and switch new assertions to Hamcrest

Addresses follow-up review feedback on PR opensearch-project#7186.

- SqsServiceTest now wraps start() in MockedStatic<SqsBackoff> and
  verifies the configured Duration reaches SqsBackoff.createExponentialBackoff.
  Previously the stub on getMaximumBackoff() was lenient and the test
  still passed if SqsService reverted to the no-arg overload.
- Converts the three new QueueConfigTest assertions to Hamcrest to match
  the project's assertion convention.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WHJ5hyg48TRqwezQFhNv8a
Signed-off-by: Luis Norman <lunorman@amazon.com>
@github-actions

Copy link
Copy Markdown

Persistent review updated to latest commit 4abb732

@srikanthpadakanti srikanthpadakanti left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM. Approving.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Make maximum back-off delay a configuration option

2 participants