add logging to track s3 records processed per object and show which object was completed when messages are deleted

Signed-off-by: Taylor Gray <tylgry@amazon.com>

Author: graytaylor0

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ObjectWorker.java

@@ -234,6 +234,8 @@ private void doProcessObject(final AcknowledgementSet acknowledgementSet,
         if (recordsWritten == 0) {
             LOG.warn("Failed to find any records in S3 object: s3ObjectReference={}.", s3ObjectReference);
             s3ObjectPluginMetrics.getS3ObjectNoRecordsFound().increment();
+        } else {
+            LOG.info("Completed reading S3 object: s3ObjectReference={}, records={}", s3ObjectReference, recordsWritten);
         }
         s3ObjectPluginMetrics.getS3ObjectSizeSummary().record(s3ObjectSize);
         s3ObjectPluginMetrics.getS3ObjectEventsSummary().record(recordsWritten);

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorker.java

@@ -290,6 +290,8 @@ && isEventBridgeEventTypeCreated(parsedMessage)) {
                             parsedMessageVisibilityTimesMap.remove(parsedMessage);
                         }
                         if (result == true) {
+                            LOG.info("Acknowledgement received, deleting SQS message for S3 object s3://{}/{}, sqsMessageId={}",
+                                parsedMessage.getBucketName(), parsedMessage.getObjectKey(), parsedMessage.getMessage().messageId());
                             final boolean successfullyDeletedAllMessages = deleteSqsMessages(waitingForAcknowledgements);
                             if (successfullyDeletedAllMessages && s3SourceConfig.isDeleteS3ObjectsOnRead()) {
                                 deleteS3Objects(s3ObjectDeletionWaitingForAcknowledgments);