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:

• aws_sqs input doesn't update the message visibility timeout while processing it #1948 — "aws_sqs input doesn't update the message visibility timeout while processing it"
• Add loop to update message visibility timeout to aws_sqs #1951 — "Add loop to update message visibility timeout to aws_sqs" (added message_timeout, refreshing the receipt handle at ~half the timeout), in internal/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 Add SQS Visibility timeout field for S3 input #3882 added nack_visibility_timeout), modeled on the heartbeat loop in internal/impl/aws/input_sqs.go.
• Key difference from aws_sqs: there it's 1 SQS message ↔ 1 in-flight message. In aws_s3 it'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.