[jimmystewpot]

Summary

This PR introduces adaptive polling concurrency ("auto-concurrency") to the aws_sqs source to optimise resource utilisation and minimise unnecessary ReceiveMessage API requests when an SQS queue is idle or experiencing low traffic.

Previously, aws_sqs initialised a fixed pool of client_concurrency polling tasks (defaulting to the number of available CPUs) at startup. Because AWS SQS is a pay-per-request service, running N concurrent long-polling loops against an idle queue resulted in N continuous ReceiveMessage calls every poll_secs interval when a single polling task would suffice.

With this change:

• Dynamic Concurrency Scaling (1..=max_concurrency): SqsSource starts with a single active polling worker and dynamically scales up to max_concurrency (defaulting to crate::num_threads()) when all active workers observe busy batches (>= 5 messages, half of MAX_BATCH_SIZE = 10 to account for standard SQS partition distribution).
• Automatic Scale-Down & Error Backoff: When a worker receives an empty poll (0 messages) or a ReceiveMessage error, surplus workers (concurrency > 1) scale down and exit atomically while guaranteeing the floor worker (concurrency == 1) never exits on idle polls. If the floor worker encounters a ReceiveError, it backs off for 1s (RECEIVE_ERROR_RETRY_DELAY) inside a cancellable select! loop to avoid tight retry loops on persistent errors.
• Immediate Sibling Cancellation on Downstream Closure: Uses a tokio::sync::watch channel to broadcast downstream SourceSender closure (PollOutcome::StreamClosed) so sibling workers blocked in long polls abort immediately rather than waiting for poll_secs to elapse.
• Configuration & Backwards Compatibility: Adds max_concurrency to AwsSqsConfig and deprecates client_concurrency with a runtime deprecation warning and fallback when client_concurrency is still configured.
• Panic-Free Hot Path: Replaces .expect("invalid timestamp") in get_timestamp with .single() so out-of-range millisecond SentTimestamp attributes return None rather than panicking, and matches (finalizer, batch_receiver) as a tuple in run_once.

References

Closes: #13322
Related: #13240

Vector configuration

sources:
  my_sqs:
    type: aws_sqs
    region: us-east-1
    queue_url: https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue
    max_concurrency: 8
    poll_secs: 15

How did you test this PR?

• Unit tests (src/sources/aws_sqs/config.rs & src/sources/aws_sqs/source.rs):
  • Added 5 configuration unit tests verifying resolve_max_concurrency() precedence (max_concurrency > deprecated client_concurrency > crate::num_threads()) and YAML rejection of max_concurrency: 0.
  • Added 9 deterministic virtual-time unit tests (#[tokio::test(start_paused = true)]) covering single-worker startup, scale-up to max_concurrency on busy batches, steady-state behaviour on partial batches (1..=4 messages), scale-down to 1 on empty polls, floor-worker backoff on ReceiveError, re-scaling after scale-down, immediate cancellation of in-flight sibling workers on StreamClosed, max_concurrency = 1 upper-bound enforcement, and worker panic propagation.
  • Added unit test verifying get_timestamp safely ignores non-numeric and out-of-range (i64::MAX) timestamps.
  • Executed cargo nextest run --no-default-features --features sources-aws_sqs -E 'test(sources::aws_sqs)' (19/19 passing).
• Integration tests (src/sources/aws_sqs/integration_tests.rs):
  • Added test_max_concurrency_delivers_multiple_batches to verify multi-batch ingestion with max_concurrency configured.
• Linting & Documentation validation:
  • FEATURES="sources-aws_sqs,aws-sqs-integration-tests" make check-clippy
  • make check-fmt
  • make check-generated-docs
  • make check-changelog-fragments
  • make check-markdown

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

Contributor Guidelines

• Please read our Vector contributor resources.
• Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
• Before pushing, follow our pre-push guidance.
• After a review is requested, please avoid force pushes to help us review incrementally.
  • Feel free to push as many commits as you want. They will be squashed into one before merging.
  • For example, you can run git merge origin master and git push.

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	⚠️ Failed 2026-10-28T10:09:59.668071Z We couldn’t complete this review. Comment @codex review to try again.	e30f623	Draft marked ready

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[Copilot]

🔵 Needs a closer look

The changes affect concurrent SQS worker lifecycle, cancellation, adaptive scaling, and acknowledgement behavior, requiring runtime and integration validation for race conditions and shutdown under real backpressure.

No open issues 🎉

What changed in this PR

This pull request adds adaptive polling concurrency to the aws_sqs source, reducing unnecessary polling for idle queues while scaling workers based on queue activity.

Changes:

• Added dynamic worker scaling, idle/error scale-down, retry backoff, and sibling cancellation.
• Added max_concurrency with deprecated client_concurrency compatibility.
• Added unit and integration coverage, safer timestamp parsing, updated documentation, and a changelog entry.

File	Description
website/​cue/​reference/​components/​sources/​generated/​aws_sqs.cue	Updates generated configuration documentation for the new and deprecated concurrency options.
website/​cue/​reference/​components/​sources/​aws_sqs.cue	Documents automatic concurrency management behavior.
src/​sources/​aws_sqs/​source.rs	Implements adaptive polling workers, cancellation, backoff, timestamp handling, and related tests.
src/​sources/​aws_sqs/​integration_tests.rs	Adds multi-batch ingestion coverage with configured maximum concurrency.
src/​sources/​aws_sqs/​config.rs	Adds concurrency configuration, deprecation handling, resolution logic, and validation tests.
changelog.d/​13322_aws_sqs_auto_concurrency.enhancement.md	Records the user-facing adaptive concurrency enhancement.

_{Review effort: Lite}

[datadoghq-integration]

More details

Static inspection found no actionable defects in adaptive polling, configuration fallback, or sibling cancellation. Runtime behavior remains unverified because tests were inspected but not executed.

Was this helpful? React 👍 or 👎

_{Open Bits AI session}

_{🤖 Bits Code Review · Commit e30f623}