Added support for multiple workers in S3 Scan Source #4439

kkondaka · 2024-04-18T16:05:51Z

Description

Added support for multiple workers in S3 Scan Source

Issues Resolved

Check List

[ X] New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
[X ] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

graytaylor0 · 2024-04-18T17:07:08Z

...api/src/main/java/org/opensearch/dataprepper/model/source/coordinator/SourceCoordinator.java

+     * @throws org.opensearch.dataprepper.model.source.coordinator.exceptions.PartitionUpdateException if the partition could not be given up due to some failure
+     * @since 2.8
+     */
+    void giveUpPartitions(SourcePartition<T> sourcePartition);


Let's name this giveUpPartition instead of giveUpPartitions

Is it a bit redundant with closePartition?

closePartition does not make partitions available (i.e. UNASSIGNED)

graytaylor0 · 2024-04-18T17:09:05Z

...src/main/java/org/opensearch/dataprepper/sourcecoordination/LeaseBasedSourceCoordinator.java

+        Optional<SourcePartitionStoreItem> ownedPartitions = Optional.empty();
+        try {
+          if (lock.tryLock()) {
+            ownedPartitions = sourceCoordinationStore.tryAcquireAvailablePartition(sourceIdentifierWithPartitionType, ownerId, DEFAULT_LEASE_TIMEOUT);


We should not put a lock around this line. We only want to lock on acquiring the global partition.

So it should look like this

Optional<SourcePartitionStoreItem> ownedPartitions = sourceCoordinationStore.tryAcquireAvailablePartition(sourceIdentifierWithPartitionType, ownerId, DEFAULT_LEASE_TIMEOUT); try { if (ownedPartitions.isEmpty && lock.tryLock()) { ... code to acquire the global partition and run supplier ... } } finally { if (lock.isHeldByCurrentThread()) { lock.unlock(); } if (ownedPartition.isEmpty()) { ownedPartitions = sourceCoordinationStore.tryAcquireAvailablePartition(sourceIdentifierWithPartitionType, ownerId, DEFAULT_LEASE_TIMEOUT); } }

graytaylor0 · 2024-04-18T17:11:59Z

...src/main/java/org/opensearch/dataprepper/sourcecoordination/LeaseBasedSourceCoordinator.java

            }
-
-            ownedPartitions = sourceCoordinationStore.tryAcquireAvailablePartition(sourceIdentifierWithPartitionType, ownerId, DEFAULT_LEASE_TIMEOUT);


This doesn't need to be in the lock either

graytaylor0 · 2024-04-18T17:20:02Z

...src/main/java/org/opensearch/dataprepper/sourcecoordination/LeaseBasedSourceCoordinator.java

        if (!initialized) {
            return;
        }
-
        final Optional<SourcePartition<T>> activePartition = partitionManager.getActivePartition();


This won't have a partition since you changed the code to not track partition manager. There are two options here

Have partition manager track all the partitions that threads are working on, and call giveUpPartition on each

Remove the PartitionManager class completely, and then on shutdown have threads stop and give up their individual partition

I would say option 2 is easier

graytaylor0 · 2024-04-18T17:20:39Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/DynamoDBSource.java


        clientFactory = new ClientFactory(awsCredentialsSupplier, sourceConfig.getAwsAuthenticationConfig(), sourceConfig.getTableConfigs().get(0).getExportConfig());
    }

+    @Override
+    public boolean areAcknowledgementsEnabled() {


You need to pull from main and rebase your branch against main to get rid of this old commit. If it still shows up then create a new branch off main and cherry-pick the commit for the worker change to it

graytaylor0 · 2024-04-18T17:21:31Z

...gins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanService.java

@@ -36,7 +42,10 @@ public class S3ScanService {
    private final AcknowledgementSetManager acknowledgementSetManager;
    private final S3ObjectDeleteWorker s3ObjectDeleteWorker;
    private final PluginMetrics pluginMetrics;
-    private ScanObjectWorker scanObjectWorker;
+    //private ScanObjectWorker scanObjectWorker;


Extra comment

graytaylor0 · 2024-04-18T17:22:51Z

...gins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanService.java

    }

    public void stop() {
-        scanObjectWorker.stop();
+        executorService.shutdown();


Let's make sure the scan workers are shutdown properly and that they give up their partition if they own one

graytaylor0 · 2024-04-18T17:25:50Z

...gins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanService.java

+    //private ScanObjectWorker scanObjectWorker;
+    private final ExecutorService executorService;
+    static final int STANDARD_BACKOFF_MILLIS = 30_000;
+    static final int FAST_BACKOFF_MILLIS = 100;


This is really fast and doesn't really make sense. Let's just make this configurable at the source and keep the default at 30 seconds. Something like

source: s3: scan: backoff_time: "1s" // default to 30 seconds

graytaylor0 · 2024-04-18T17:28:08Z

Also, there are a lot of failing LeaseBasedSourceCoordinator tests in the build

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0

Overall looks good just had some comments

graytaylor0 · 2024-04-22T23:18:40Z

...api/src/main/java/org/opensearch/dataprepper/model/source/coordinator/SourceCoordinator.java

     */
-    void giveUpPartitions();
+    void giveUpPartition(SourcePartition<T> sourcePartition);


This can just take a String partitionKey to align with the other functions here

graytaylor0 · 2024-04-22T23:20:41Z

...src/main/java/org/opensearch/dataprepper/sourcecoordination/LeaseBasedSourceCoordinator.java

-        }
+        Optional<SourcePartitionStoreItem> ownedPartitions = Optional.empty();
+        try {
+          if (lock.tryLock()) {


This lock is still blocking the tryAcquireAvailablePartition call unnecessarily. Please change to align with my comment below.

graytaylor0 · 2024-04-22T23:21:18Z

...src/main/java/org/opensearch/dataprepper/sourcecoordination/LeaseBasedSourceCoordinator.java

    }

    @Override
-    public void giveUpPartitions() {
+    public void giveUpPartition(SourcePartition<T> sourcePartition) {


You only need the String partitionKey here

graytaylor0 · 2024-04-22T23:27:40Z

...s/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ScanObjectWorker.java

@@ -90,6 +93,7 @@ public ScanObjectWorker(final S3Client s3Client,
        this.pluginMetrics = pluginMetrics;
        acknowledgementSetCallbackCounter = pluginMetrics.counter(ACKNOWLEDGEMENT_SET_CALLBACK_METRIC_NAME);
        this.sourceCoordinator.initialize();
+        this.partitions = new ArrayList<>();

        this.partitionCreationSupplier = new S3ScanPartitionCreationSupplier(s3Client, bucketOwnerProvider, scanOptionsBuilderList, s3ScanSchedulingOptions);


Nit: This class can be created only once and share between all the threads. Having it as it is know won't cause any issues though, just some extra instantiations.

You mean the ScanObjectWorker class? We need a Runnable class for each thread. Not using this means we need to define another runnable class, right?

No the S3ScanPartitionCreationSupplier class. We could only instantiate it once and pass to all of the ScanObjectWorkers

graytaylor0 · 2024-04-22T23:28:48Z

...s/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ScanObjectWorker.java

                    }
                }, ACKNOWLEDGEMENT_SET_TIMEOUT);
+                partitions.remove(objectToProcess.get());


I think this is meant to be in the callback?

Yes. Thanks for catching it.

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 · 2024-04-23T14:41:25Z

...src/main/java/org/opensearch/dataprepper/sourcecoordination/LeaseBasedSourceCoordinator.java

@@ -450,6 +439,6 @@ private void giveUpAndSaveGlobalStateForPartitionCreation(final SourcePartitionS
    private SourcePartitionStoreItem getItemWithAction(final String partitionKey, final String action, final Boolean fromAcknowledgmentsCallback) {
        // The validation against activePartition in partition manager needs to be skipped when called from acknowledgments callback
        // because otherwise it will fail the validation since it is actively working on a different partition when ack is received
-        return fromAcknowledgmentsCallback ? getSourcePartitionStoreItem(partitionKey, action) : validateAndGetSourcePartitionStoreItem(partitionKey, action);
+        return fromAcknowledgmentsCallback ? getSourcePartitionStoreItem(partitionKey, action) : getSourcePartitionStoreItem(partitionKey, action);


Nit: No need for this ternary operator anymore.

Oh, right. Will change it.

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

…ect#4439) * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> * Fixed failing integration tests Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> * Fixed failing check style Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> --------- Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com> Co-authored-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

kkondaka requested review from chenqi0805, engechas, graytaylor0, dinujoh, asifsmohammed, KarstenSchnitter, dlvenable and oeyh as code owners April 18, 2024 16:05

graytaylor0 reviewed Apr 18, 2024

View reviewed changes

kkondaka force-pushed the s3-scan-workers branch from da4f0a5 to a2961af Compare April 22, 2024 19:55

Addressed review comments

e0363e3

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

kkondaka force-pushed the s3-scan-workers branch from a2961af to e0363e3 Compare April 22, 2024 21:56

Krishna Kondaka added 2 commits April 22, 2024 22:30

Fixed failing integration tests

abe3d70

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed failing check style

ec1b8b0

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 reviewed Apr 22, 2024

View reviewed changes

Addressed review comments

e0be8d7

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 reviewed Apr 23, 2024

View reviewed changes

graytaylor0 previously approved these changes Apr 23, 2024

View reviewed changes

Addressed review comments

68cda78

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

kkondaka dismissed graytaylor0’s stale review via 68cda78 April 23, 2024 15:14

graytaylor0 approved these changes Apr 23, 2024

View reviewed changes

chenqi0805 approved these changes Apr 23, 2024

View reviewed changes

kkondaka merged commit 3cc41b7 into opensearch-project:main Apr 23, 2024
66 of 68 checks passed

kkondaka added this to the v2.8 milestone May 14, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added support for multiple workers in S3 Scan Source #4439

Added support for multiple workers in S3 Scan Source #4439

kkondaka commented Apr 18, 2024

graytaylor0 Apr 18, 2024

chenqi0805 Apr 18, 2024

graytaylor0 Apr 23, 2024

graytaylor0 Apr 18, 2024

graytaylor0 Apr 18, 2024

graytaylor0 Apr 18, 2024

graytaylor0 Apr 18, 2024

graytaylor0 Apr 18, 2024

graytaylor0 Apr 18, 2024

graytaylor0 Apr 18, 2024

graytaylor0 commented Apr 18, 2024

graytaylor0 left a comment

graytaylor0 Apr 22, 2024

graytaylor0 Apr 22, 2024

graytaylor0 Apr 22, 2024

graytaylor0 Apr 22, 2024

kkondaka Apr 23, 2024

graytaylor0 Apr 23, 2024 •

edited

Loading

graytaylor0 Apr 22, 2024

kkondaka Apr 23, 2024

graytaylor0 Apr 23, 2024

kkondaka Apr 23, 2024

		}

		ownedPartitions = sourceCoordinationStore.tryAcquireAvailablePartition(sourceIdentifierWithPartitionType, ownerId, DEFAULT_LEASE_TIMEOUT);

Added support for multiple workers in S3 Scan Source #4439

Added support for multiple workers in S3 Scan Source #4439

Conversation

kkondaka commented Apr 18, 2024

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 commented Apr 18, 2024

graytaylor0 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 Apr 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 Apr 23, 2024 •

edited

Loading