redpanda-data / redpanda-data/connect
aws_s3 input: support SQS visibility-timeout heartbeat (message_timeout) to avoid reprocessing slow/large objects
Nobody has claimed this yet.
- Dominant language
- Go
- Stars
- 8.8k
- Forks
- 969
- Avg merge
- 1d 13h
- Merged PRs (30d)
- 64
Description
Problem
The aws_s3 input (SQS-driven mode) has no mechanism to extend the visibility timeout of an in-flight SQS notification while its S3 object is being processed. The SQS message for an object is deleted only after all records scanned from it are acked downstream, but nothing keeps the message hidden in the meantime. If download + scan + downstream delivery exceeds the queue's visibility timeout, SQS redelivers the notification and the entire object is processed again → duplicate records.
This is acknowledged in the docs ("if the S3 object takes longer to process than the visibility timeout of your queue, then the same objects might be processed multiple times"), but there's currently no way to opt into standard SQS consumer behavior (heartbeat the receipt handle during processing).
It's especially easy to hit when the downstream output batches on a time period: a small object can take up to the full batch period to ack, independent of object size, so even fast objects get redelivered under a typical 30s queue visibility timeout.
Current behavior (verified on redpanda-connect 4.92.0)
The aws_s3.sqs block exposes only:
url, endpoint, key_path, bucket_path, envelope_path, delay_period, max_messages, wait_time_seconds, nack_visibility_timeout
There is no message_timeout/heartbeat field. nack_visibility_timeout (added in #3882) only sets the visibility applied on nack — its own PR description notes it is "for nack handling (not per-message)" — so it does not address in-flight processing.
Composing existing components doesn't solve it either: there's no S3 "get object" processor to pair with the aws_sqs input, and even if there were, doing the fetch in a processor would lose the input's streaming decompress/json_documents scanner and require loading whole objects into memory.
Precedent
The sibling aws_sqs input already solved the identical problem:
- #1948 — "aws_sqs input doesn't update the message visibility timeout while processing it"
- #1951 — "Add loop to update message visibility timeout to aws_sqs" (added
message_timeout, refreshing the receipt handle at ~half the timeout), ininternal/impl/aws/input_sqs.go.
Proposed solution
Add a message_timeout field to the aws_s3.sqs block, mirroring aws_sqs, that periodically calls ChangeMessageVisibility on the in-flight notification until the object is fully acked (or nacked). Default it to preserve today's behavior (off / current default) so existing deployments are unaffected.
Implementation notes
- Target:
internal/impl/aws/input_s3.go(where #3882 addednack_visibility_timeout), modeled on the heartbeat loop ininternal/impl/aws/input_sqs.go. - Key difference from
aws_sqs: there it's 1 SQS message ↔ 1 in-flight message. Inaws_s3it's 1 notification ↔ N records (every scanned line), with the SQS delete deferred until all N resolve. So the heartbeat must run for the object's whole processing lifetime, keyed to the notification's receipt handle, and stop when the last derived record is acked/nacked — reusing the input's existing per-object ack tracking.
Backward compatibility
New field, default preserves current behavior. No change for users who don't set it.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start in internal/impl/aws/input_s3.go, especially the existing nack_visibility_timeout handling and per-object acknowledgement tracking. Compare the heartbeat loop in internal/impl/aws/input_sqs.go and trace how one SQS notification maps to multiple scanned records. Done means a new optional message_timeout setting keeps the receipt handle visible through the object's processing lifetime while preserving current behavior when unset.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, go
- Domain
- data-engineering, stream-processing
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 68/100