Change s3 scan and opensearch to only save state every 5 minutes, fix… #3581
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
graytaylor0
requested review from
chenqi0805,
engechas,
dinujoh,
kkondaka,
asifsmohammed,
dlvenable and
oeyh
as code owners
graytaylor0
force-pushed
the
FixCheckpointRateForOSSourceAndS3Scan
branch
from
9c2658a
to
0bf0917
Compare
… bug where any action was valid in OpenSearch sink Signed-off-by: Taylor Gray <tylgry@amazon.com>
graytaylor0
force-pushed
the
FixCheckpointRateForOSSourceAndS3Scan
branch
from
0bf0917
to
823da1e
Compare
dinujoh
reviewed
Nov 3, 2023
| if (sourceCoordinator != null && partitionKey != null && | ||
| (System.currentTimeMillis() - lastCheckpointTime.get() > DEFAULT_CHECKPOINT_INTERVAL_MILLS)) { | ||
| LOG.debug("Renew partition ownership for the object {}", partitionKey); | ||
| lastCheckpointTime.set(System.currentTimeMillis()); |
There was a problem hiding this comment.
should this be set after saveProgressState is called ? What happens in the saveProgressState api fails ?
There was a problem hiding this comment.
Good call will swap the order
Comment on lines
-191
to
-192
| // Should do one last checkpoint when done. | ||
| verify(coordinator).saveProgressStateForPartition(any(DataFilePartition.class), eq(null)); |
There was a problem hiding this comment.
why is this removed ? Is there a way to validate it ?
There was a problem hiding this comment.
The coordinator class is not use in the DataFileLoader. This check was here because we weren't mocking the checkpointer class. Now we are mocking it so the behavior is tested in the Checkpointer unit tests.
Signed-off-by: Taylor Gray <tylgry@amazon.com>
dinujoh
approved these changes
Nov 3, 2023
kkondaka
approved these changes
Nov 5, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
… bug where any action was valid in OpenSearch sink
Signed-off-by: Taylor Gray tylgry@amazon.com
Description
saveProgressStateis called to the coordination store for partitions in s3 scan and the opensearch source. Before they were calling saveState every 10,000 Events, which was far too little. It is better to use a time based measurementThey were calling it too often, leading to wasted writes on the store. The default lease timeout is 10 minutes, so we have them call every 5 minutes.
Check List
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.