perf(plugin): optimize aws s3 file listing

Lower number of requests sent to S3 while listing objects. For buckets
with large number of objects, listing can take some time, because there
is a request for metadata sent to s3 for each object in the bucket.
This is redundant, because all data needed for objects listing are
available in S3ObjectSummary. This reduce listing time significantly.

Resolves: #490

Author: bjarosze

connect-file-pulse-filesystems/filepulse-amazons3-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/AmazonS3FileSystemListing.java

@@ -95,11 +95,6 @@ public Collection<FileObjectMeta> listObjects() {
 
                 objectMetaList.addAll(objectListing.getObjectSummaries()
                         .stream()
-                        .map(s3ObjectSummary ->
-                                new S3BucketKey(
-                                        s3ObjectSummary.getBucketName(),
-                                        s3ObjectSummary.getKey()
-                                ))
                         .map(s3Storage::getObjectMetadata)
                         .filter(Objects::nonNull)
                         .collect(Collectors.toList()));

connect-file-pulse-filesystems/filepulse-amazons3-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/AmazonS3Storage.java

@@ -27,6 +27,7 @@
 import com.amazonaws.services.s3.model.GetObjectMetadataRequest;
 import com.amazonaws.services.s3.model.GetObjectRequest;
 import com.amazonaws.services.s3.model.ObjectMetadata;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
 import com.amazonaws.services.s3.model.StorageClass;
 import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
 import io.streamthoughts.kafka.connect.filepulse.source.GenericFileObjectMeta;
@@ -186,6 +187,13 @@ public FileObjectMeta getObjectMetadata(final S3BucketKey s3Object) {
         );
     }
 
+    public FileObjectMeta getObjectMetadata(final S3ObjectSummary s3ObjectSummary) {
+        return createFileObjectMeta(
+                new S3BucketKey(s3ObjectSummary.getBucketName(), s3ObjectSummary.getKey()),
+                s3ObjectSummary
+        );
+    }
+
     private ObjectMetadata loadObjectMetadata(final S3BucketKey s3Object) {
         var request = new GetObjectMetadataRequest(s3Object.bucketName(), s3Object.key());
         try {
@@ -264,11 +272,11 @@ private static FileObjectMeta createFileObjectMeta(final S3BucketKey s3Object,
         userDefinedMetadata.put("s3.object.summary.etag", objectMetadata.getETag());
         userDefinedMetadata.put("s3.object.summary.storageClass", objectMetadata.getStorageClass());
 
-        final String contentMD5 = objectMetadata.getContentMD5();
+        final String contentMD5 = objectMetadata.getETag();
 
         FileObjectMeta.ContentDigest digest = null;
         if (contentMD5 != null) {
-            digest = new FileObjectMeta.ContentDigest(contentMD5, "MD5");
+            digest = new FileObjectMeta.ContentDigest(contentMD5, "ETAG");
         }
 
         return new GenericFileObjectMeta.Builder()
@@ -280,4 +288,22 @@ private static FileObjectMeta createFileObjectMeta(final S3BucketKey s3Object,
                 .withUserDefinedMetadata(userDefinedMetadata)
                 .build();
     }
+
+    private static FileObjectMeta createFileObjectMeta(final S3BucketKey s3Object,
+                                                       final S3ObjectSummary s3ObjectSummary) {
+        final String contentMD5 = s3ObjectSummary.getETag();
+
+        FileObjectMeta.ContentDigest digest = null;
+        if (contentMD5 != null) {
+            digest = new FileObjectMeta.ContentDigest(contentMD5, "ETAG");
+        }
+
+        return new GenericFileObjectMeta.Builder()
+                .withUri(s3Object.toURI())
+                .withName(s3ObjectSummary.getKey())
+                .withContentLength(s3ObjectSummary.getSize())
+                .withLastModified(s3ObjectSummary.getLastModified())
+                .withContentDigest(digest)
+                .build();
+    }
 }