Add support for fully async acknowledgments in source coordination

[graytaylor0]

Description

This changes adds a new function to the source coordinator interface

updatePartitionForAckWait(final String partitionKey, final Duration acknowledgmentTimeout)

This method will give ownership of acknowledgmentTimeout amount for the acknowledgment to be received. If the acknowledgment for the partition is not received in time, then another instance of Data Prepper will attempt to process it from the beginning. Calling the updatePartitionForAckWait is also necessary to allow the current node to pick up a new partition to process immediately.

Also added a paramter fromAcknowledgmentCallback to close/completePartition. Only the callback functions should set this to true when the partition is closed or completed by the callback.

• Implemented the use of this change in both s3 scan and the opensearch source to have acknowledgment timeout of 2 hours (we could consider making this configurable in the future, but given that we have infinite timeout in Kafka, I was hoping we would consolidate on the user experience for acknowledgments timeout first) Instead of waiting to receive the acknowledgments from the sink, it will now be done completely asynchronously, meaning the source will continuously pick up the next partition and write it to the buffer.

Issues Resolved

Resolves #3381

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• 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]

Let's use full words in interface names. updatePartitionForAcknowledgementWait.

[dlvenable]

Let's make "complete" a constant now that you are using it twice.

[dlvenable]

Is there an object we can use here instead? Is there any acknowledgements handle of any sort? This would be ideal so that if data is needed from there in the future, we don't have to make yet another change to the interface.

[graytaylor0]

I don't think there's anything that makes sense to pass for acknowledgments other than the boolean. As far as objects in requests, it could make sense to keep the model class the same with

void completePartition(final CompletePartitionRequest)

and to do that with the rest of the SourceCoordinator methods as well. I'm not sure if that's worth a large refactor though

[dlvenable]

I only suggested it if there is an existing model. I'm good keeping this as it is.

[dlvenable]

This logic is very similar to what you have above in another method. It would be best to move this into a private method so that future maintainers/developers don't miss the other location when modifying the code.

[kkondaka]

Can you add a comment on why validation is not needed if this is called from acknowledgement callback?

[opensearch-trigger-bot]

The backport to 2.4 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.4 2.4
# Navigate to the new working tree
cd .worktrees/backport-2.4
# Create a new branch
git switch --create backport/backport-3384-to-2.4
# Cherry-pick the merged commit of this pull request and resolve the conflicts
git cherry-pick -x --mainline 1 6546901a8492500adda6d4fcfbc7d7f014734d47
# Push it to GitHub
git push --set-upstream origin backport/backport-3384-to-2.4
# Go back to the original working tree
cd ../..
# Delete the working tree
git worktree remove .worktrees/backport-2.4

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