[BUG] Visibility duplication protection fails when using S3 source for large files and receiving 10 messages from SQS queue

[danhli]

Describe the bug
When ingesting large files using S3-SQS processing, OpenSearch Ingestion pipeline failed to prevent duplication although visibility_duplication_protection was set to true. The SQS queue's ApproximateNumberOfMessagesVisible metric kept growing. Confirmed duplicated documents existed in the OpenSearch index. There were many errors in the pipeline logs. For example:
ERROR org.opensearch.dataprepper.plugins.source.s3.SqsWorker - Failed to set visibility timeout for message [foo] to 60
software.amazon.awssdk.services.sqs.model.SqsException: Value [bar] for parameter ReceiptHandle is invalid. Reason: Message does not exist or is not available for visibility timeout change. (Service: Sqs, Status Code: 400, Request ID: [baz])

The issue happened when OSI received 10 messages (or close to 10) from the SQS queue in a single request. It worked fine when OSI received one message in a single request.

To Reproduce
Steps to reproduce the behavior:

1. Configure a S3 bucket and a SQS queue according to the documentation
2. Set the following parameters in the pipeline configuration
   source:
   s3:
   acknowledgments: true
   notification_type: "sqs"
   compression: "gzip"
   codec:
   ndjson:
   workers: 5
   sqs:
   queue_url:
   maximum_messages: 10
   visibility_timeout: "60s"
   visibility_duplication_protection: true
3. Provision the right amount of OCUs for the pipeline
4. Start the pipeline
5. Send large gzipped files, e.g. 80MB, to the S3 bucket continuously
6. Check the pipeline's logs using CloudWatch Log Insights with the following query
   fields @timestamp, @message
   | filter @message like /ReceiptHandle is invalid/
   | sort @timestamp desc
7. Check the pipeline's logs using CloudWatch Log Insights with the following query
   fields @timestamp, @message
   | filter @message like /10 messages from SQS. Processing/
   | sort @timestamp desc
8. Check the queue's ApproximateNumberOfMessagesVisible metric
9. Check OpenSearch index for duplicated documents

Expected behavior
No log messages are found from step 6. Log messages are found from step 7. The ApproximateNumberOfMessagesVisible metric doesn't keep growing. There are no duplicated documents in the index.