Add ProgressCheck callbacks to end-to-end acknowledgements #3565

kkondaka · 2023-10-31T15:17:01Z

Description

Add ProgressCheck callbacks to end-to-end acknowledgements
Adds optional progress check callbacks the acknowledgement set create api

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

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.

dlvenable · 2023-11-01T16:16:44Z

...c/main/java/org/opensearch/dataprepper/model/acknowledgements/AcknowledgementSetManager.java

@@ -30,6 +30,7 @@ public interface AcknowledgementSetManager {
     */
    AcknowledgementSet create(final Consumer<Boolean> callback, final Duration timeout);

+    AcknowledgementSet create(final Consumer<Double> progressCheckCallback, final Duration progressCheckInterval, final Consumer<Boolean> callback, final Duration timeout);


Since the progress check is optional, it might make sense to add the progress check on the AcknowledgementSet itself. This will help with clarity in the caller's code.

e.g.

acknowledgementSet = manager.create(myCallback, timeout); acknowledgementSet.addProgressCheck(myProgressCallback, progressInterval);

We may also want an expiry callback right? So continuing to add to the create method is making the code harder to read. I'm open to other ideas as well - perhaps a builder?

manager.builder(myCallback, timeout) .withProgressCheck(myProgressCallback, progressInterval) .build()

Because progress check is optional, we have two different constructors. Isn't that good alternative way?

What if I want an expiry callback, but not progress? This becomes confusing to use. It is also already difficult to read and understand in the SQS code because you need to know the order of these callbacks and durations.

dlvenable · 2023-11-01T16:21:14Z

.../main/java/org/opensearch/dataprepper/acknowledgements/DefaultAcknowledgementSetManager.java

@@ -36,12 +39,23 @@ public DefaultAcknowledgementSetManager(
    public DefaultAcknowledgementSetManager(final ExecutorService callbackExecutor, final Duration waitTime) {
        this.acknowledgementSetMonitor = new AcknowledgementSetMonitor();
        this.executor = Objects.requireNonNull(callbackExecutor);
+        // Single thread executor should be sufficient in most cases (esp if the underlying host has 1 or 2 cores)
+        this.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();


The current executor is dependency injected. I think we should continue that pattern. It comes from:

data-prepper/data-prepper-core/src/main/java/org/opensearch/dataprepper/acknowledgements/AcknowledgementAppConfig.java

Lines 25 to 28 in 068c571

@Bean(name = "acknowledgementCallbackExecutor")

ExecutorService acknowledgementCallbackExecutor(final CallbackTheadFactory callbackTheadFactory) {

return Executors.newFixedThreadPool(MAX_THREADS, callbackTheadFactory);

}

Also, can we share a single executor?

Change this:

https://github.com/opensearch-project/data-prepper/blob/068c57172c0dfb0c4e43d1d796e8a59fa11fb1df/data-prepper-core/src/main/java/org/opensearch/dataprepper/acknowledgements/AcknowledgementAppConfig.java#L27C28-L27C28

To this:

return Executors.newScheduledThreadPool(MAX_THREADS, callbackTheadFactory);

And change the interfaces to use ScheduledExecutorService throughout.

dlvenable · 2023-11-01T16:23:20Z

...ore/src/main/java/org/opensearch/dataprepper/acknowledgements/DefaultAcknowledgementSet.java


-    public DefaultAcknowledgementSet(final ExecutorService executor, final Consumer<Boolean> callback, final Duration expiryTime, final DefaultAcknowledgementSetMetrics metrics) {
+    public DefaultAcknowledgementSet(final ExecutorService executor,


See my other comment below. I believe that we can use a single ScheduledExecutorService.

dlvenable · 2023-11-01T16:24:55Z

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

+                acknowledgementSet = acknowledgementSetManager.create(
+                    (ratio) -> {
+                        final int newVisibilityTimeoutSeconds = (int) (visibilityTimeout);
+                        final ChangeMessageVisibilityRequest changeMessageVisibilityRequest = ChangeMessageVisibilityRequest.builder()


I tend to think that we should make this a configuration option. It is changing existing behavior and is now requiring access to a new API and thus requires new permissions.

Also, we should let the user configure a maximum. What if the acknowledgement set is stuck for some reason and it never completes after some number of hours?

Signed-off-by: Krishna Kondaka <[email protected]>

graytaylor0 · 2023-11-03T05:21:50Z

.../java/org/opensearch/dataprepper/acknowledgements/DefaultAcknowledgementSetManagerTests.java

+    private DefaultEventHandle eventHandle1;
+    private DefaultEventHandle eventHandle2;
+    private DefaultEventHandle eventHandle3;
+    private DefaultEventHandle eventHandle4;
+    private DefaultEventHandle eventHandle5;
+    private DefaultEventHandle eventHandle6;


Nit: Could make this a List<DefaultEventHandle> just to make it more flexible for future test case additions

graytaylor0 · 2023-11-03T05:34:24Z

...rce/src/main/java/org/opensearch/dataprepper/plugins/source/s3/configuration/SqsOptions.java

+    @JsonProperty("extend_visibility_timeout")
+    private Boolean extendVisibilityTimeout = DEFAULT_EXTEND_VISIBILITY_TIMEOUT;
+
+    @JsonProperty("max_visibility_timeout_extesion")


typo here on extension

dlvenable

I have a few more comments to consider. Also, let's correct the property misspelling that @graytaylor0 noted.

dlvenable · 2023-11-03T14:35:34Z

...ource/src/integrationTest/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorkerIT.java

+                events.add(event);
+            }
+            try {
+                Thread.sleep(4000);


Can we decrease this sleep?

dlvenable · 2023-11-03T14:40:24Z

...-api/src/main/java/org/opensearch/dataprepper/model/acknowledgements/AcknowledgementSet.java

+     * @param progressCheckInterval frequency of invocation of progress check callback
+     * @since 2.6
+     */
+    public void addProgressCheck(final Consumer<Double> progressCheckCallback, final Duration progressCheckInterval);


I'm curious why you chose to make this a Consumer<Double>. The current code isn't using this ratio presently.

A few alternatives:

Make this a Runnable without any input.

Create a new interface that can be provided so that the relevant arguments can be expanded in the future. For example, Consumer<ProgressCheck>.

public interface ProgressCheck { Double getRatio(); }

Then we could add other items if desired such as Duration getTimeToExpiry().

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable

There are checkstyle violations in this PR. I'm good once the build is running.

dlvenable · 2023-11-03T20:32:57Z

[ERROR] /home/runner/work/data-prepper/data-prepper/data-prepper-plugins/s3-source/src/integrationTest/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorkerIT.java:41:8: Unused import - java.util.concurrent.atomic.AtomicInteger. [UnusedImports]

Signed-off-by: Krishna Kondaka <[email protected]>

opensearch-trigger-bot · 2023-11-05T18:41:42Z

The backport to 2.5 failed:

The process '/usr/bin/git' failed with exit code 1

To backport manually, run these commands in your terminal:

# Fetch latest updates from GitHub
git fetch
# Create a new working tree
git worktree add .worktrees/backport-2.5 2.5
# Navigate to the new working tree
cd .worktrees/backport-2.5
# Create a new branch
git switch --create backport/backport-3565-to-2.5
# Cherry-pick the merged commit of this pull request and resolve the conflicts
git cherry-pick -x --mainline 1 6681e75d8b8cfa3985e4a11f9fa9d6238562e462
# Push it to GitHub
git push --set-upstream origin backport/backport-3565-to-2.5
# Go back to the original working tree
cd ../..
# Delete the working tree
git worktree remove .worktrees/backport-2.5

Then, create a pull request where the base branch is 2.5 and the compare/head branch is backport/backport-3565-to-2.5.

dlvenable · 2023-11-06T13:17:23Z

@asifsmohammed , This should not be back ported as it is a new feature.

dlvenable requested changes Nov 1, 2023

View reviewed changes

Add ProgressCheck callbacks to end-to-end acknowledgements

21f3526

Signed-off-by: Krishna Kondaka <[email protected]>

kkondaka force-pushed the e2eack-changes branch from c486f01 to 21f3526 Compare November 2, 2023 06:33

kkondaka marked this pull request as ready for review November 2, 2023 06:33

kkondaka requested review from chenqi0805, engechas, graytaylor0, dinujoh, asifsmohammed and oeyh as code owners November 2, 2023 06:33

Krishna Kondaka added 4 commits November 2, 2023 15:11

Removed unnecessary logs

5a9ab5c

Signed-off-by: Krishna Kondaka <[email protected]>

Fixed unnecessary change

057b9d1

Signed-off-by: Krishna Kondaka <[email protected]>

Fixed build failure in KafkaSourceJsonTypeIT

211ac55

Signed-off-by: Krishna Kondaka <[email protected]>

Fix sqs worker integration test failure

e8455cf

Signed-off-by: Krishna Kondaka <[email protected]>

graytaylor0 reviewed Nov 3, 2023

View reviewed changes

dlvenable requested changes Nov 3, 2023

View reviewed changes

Krishna Kondaka added 2 commits November 3, 2023 19:06

Addressed review comments

740cb4b

Signed-off-by: Krishna Kondaka <[email protected]>

Added Progress Check interface and test files

d3bfac4

Signed-off-by: Krishna Kondaka <[email protected]>

graytaylor0 previously approved these changes Nov 3, 2023

View reviewed changes

dlvenable approved these changes Nov 3, 2023

View reviewed changes

dlvenable requested changes Nov 3, 2023

View reviewed changes

Addressed review comments

cf6e8f8

Signed-off-by: Krishna Kondaka <[email protected]>

kkondaka dismissed graytaylor0’s stale review via cf6e8f8 November 3, 2023 20:47

Addressed review comments

1b8fad6

Signed-off-by: Krishna Kondaka <[email protected]>

graytaylor0 approved these changes Nov 3, 2023

View reviewed changes

dlvenable approved these changes Nov 4, 2023

View reviewed changes

dlvenable merged commit 6681e75 into opensearch-project:main Nov 4, 2023
66 checks passed

asifsmohammed added the backport 2.5 label Nov 5, 2023

dlvenable removed the backport 2.5 label Nov 6, 2023

This was referenced Nov 8, 2023

Support dynamically changing the visibility timeout for S3 Source with SQS queue #2485

Closed

Visibility timeout extension configuration changes #3604

Merged

kkondaka deleted the e2eack-changes branch May 13, 2024 05:51

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add ProgressCheck callbacks to end-to-end acknowledgements #3565

Add ProgressCheck callbacks to end-to-end acknowledgements #3565

kkondaka commented Oct 31, 2023

dlvenable Nov 1, 2023

kkondaka Nov 1, 2023

dlvenable Nov 1, 2023

dlvenable Nov 1, 2023

dlvenable Nov 1, 2023

dlvenable Nov 1, 2023

graytaylor0 Nov 3, 2023

graytaylor0 Nov 3, 2023

dlvenable left a comment

dlvenable Nov 3, 2023

dlvenable Nov 3, 2023

dlvenable left a comment

dlvenable commented Nov 3, 2023

opensearch-trigger-bot bot commented Nov 5, 2023

dlvenable commented Nov 6, 2023

	@Bean(name = "acknowledgementCallbackExecutor")
	ExecutorService acknowledgementCallbackExecutor(final CallbackTheadFactory callbackTheadFactory) {
	return Executors.newFixedThreadPool(MAX_THREADS, callbackTheadFactory);
	}


		public DefaultAcknowledgementSet(final ExecutorService executor, final Consumer<Boolean> callback, final Duration expiryTime, final DefaultAcknowledgementSetMetrics metrics) {
		public DefaultAcknowledgementSet(final ExecutorService executor,

Add ProgressCheck callbacks to end-to-end acknowledgements #3565

Add ProgressCheck callbacks to end-to-end acknowledgements #3565

Conversation

kkondaka commented Oct 31, 2023

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

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

dlvenable commented Nov 3, 2023

opensearch-trigger-bot bot commented Nov 5, 2023

dlvenable commented Nov 6, 2023