Do not filter out objects based on last modified timestamp when delet…

…e_objects_on_read is enabled Signed-off-by: Taylor Gray <[email protected]>
opensearch-project · Jan 9, 2025 · 44f67ae · 44f67ae
1 parent f264c09
commit 44f67ae
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 5 deletions.
diff --git a/...in/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanPartitionCreationSupplier.java b/...in/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanPartitionCreationSupplier.java
@@ -5,6 +5,7 @@
 
 package org.opensearch.dataprepper.plugins.source.s3;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.opensearch.dataprepper.model.source.coordinator.PartitionIdentifier;
 import org.opensearch.dataprepper.plugins.source.s3.configuration.FolderPartitioningOptions;
 import org.opensearch.dataprepper.plugins.source.s3.configuration.S3ScanKeyPathOption;
@@ -50,17 +51,21 @@ public class S3ScanPartitionCreationSupplier implements Function<Map<String, Obj
 
     private final FolderPartitioningOptions folderPartitioningOptions;
 
+    private final boolean deleteS3ObjectsOnRead;
+
     public S3ScanPartitionCreationSupplier(final S3Client s3Client,
                                            final BucketOwnerProvider bucketOwnerProvider,
                                            final List<ScanOptions> scanOptionsList,
                                            final S3ScanSchedulingOptions schedulingOptions,
-                                           final FolderPartitioningOptions folderPartitioningOptions) {
+                                           final FolderPartitioningOptions folderPartitioningOptions,
+                                           final boolean deleteS3ObjectsOnRead) {
 
         this.s3Client = s3Client;
         this.bucketOwnerProvider = bucketOwnerProvider;
         this.scanOptionsList = scanOptionsList;
         this.schedulingOptions = schedulingOptions;
         this.folderPartitioningOptions = folderPartitioningOptions;
+        this.deleteS3ObjectsOnRead = deleteS3ObjectsOnRead;
     }
 
     @Override
@@ -186,9 +191,10 @@ private void initializeGlobalStateMap(final Map<String, Object> globalStateMap)
         globalStateMap.put(SINGLE_SCAN_COMPLETE, false);
     }
 
-    private boolean isLastModifiedTimeAfterMostRecentScanForBucket(final Instant previousScanTime,
+    @VisibleForTesting
+    boolean isLastModifiedTimeAfterMostRecentScanForBucket(final Instant previousScanTime,
                                                                    final S3Object s3Object) {
-        if (previousScanTime == null) {
+        if (previousScanTime == null || deleteS3ObjectsOnRead) {
             return true;
         }
 

diff --git a/...3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ScanObjectWorker.java b/...3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ScanObjectWorker.java
@@ -129,7 +129,7 @@ public ScanObjectWorker(final S3Client s3Client,
         this.folderPartitioningOptions = s3SourceConfig.getS3ScanScanOptions().getPartitioningOptions();
         this.acknowledgmentSetTimeout = s3SourceConfig.getS3ScanScanOptions().getAcknowledgmentTimeout();
 
-        this.partitionCreationSupplier = new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsBuilderList, s3ScanSchedulingOptions, s3SourceConfig.getS3ScanScanOptions().getPartitioningOptions());
+        this.partitionCreationSupplier = new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsBuilderList, s3ScanSchedulingOptions, s3SourceConfig.getS3ScanScanOptions().getPartitioningOptions(), s3SourceConfig.isDeleteS3ObjectsOnRead());
         this.acknowledgmentsRemainingForPartitions = new ConcurrentHashMap<>();
         this.objectsToDeleteForAcknowledgmentSets = new ConcurrentHashMap<>();
     }

diff --git a/...ava/org/opensearch/dataprepper/plugins/source/s3/S3ScanPartitionCreationSupplierTest.java b/...ava/org/opensearch/dataprepper/plugins/source/s3/S3ScanPartitionCreationSupplierTest.java
@@ -69,15 +69,18 @@ public class S3ScanPartitionCreationSupplierTest {
 
     private FolderPartitioningOptions folderPartitioningOptions;
 
+    private boolean isDeleteS3ObjectsOnRead;
+
     @BeforeEach
     void setup() {
         scanOptionsList = new ArrayList<>();
         folderPartitioningOptions = null;
+        isDeleteS3ObjectsOnRead = false;
     }
 
 
     private Function<Map<String, Object>, List<PartitionIdentifier>> createObjectUnderTest() {
-        return new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsList, schedulingOptions, folderPartitioningOptions);
+        return new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsList, schedulingOptions, folderPartitioningOptions, isDeleteS3ObjectsOnRead);
     }
 
     @Test
@@ -452,4 +455,13 @@ void getNextPartition_with_folder_partitioning_enabled_returns_the_expected_part
         assertThat(resultingPartitions.stream().map(PartitionIdentifier::getPartitionKey).collect(Collectors.toList()),
                 containsInAnyOrder(expectedPartitionIdentifiers.stream().map(PartitionIdentifier::getPartitionKey).map(Matchers::equalTo).collect(Collectors.toList())));
     }
+
+    @Test
+    void isLastModifiedTimeAfterMostRecentScanForBucket_returns_true_when_disable_s3_objects_on_read_is_enabled() {
+        isDeleteS3ObjectsOnRead = true;
+        final S3ScanPartitionCreationSupplier objectUnderTest = new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsList, schedulingOptions, folderPartitioningOptions, isDeleteS3ObjectsOnRead);
+
+        final boolean result = objectUnderTest.isLastModifiedTimeAfterMostRecentScanForBucket(Instant.now(), mock(S3Object.class));
+        assertThat(result, equalTo(true));
+    }
 }