Skip to content

Introduce checkpointing per shard to DDB source, via ShardAcknowledgementManager class#5818

Merged
graytaylor0 merged 14 commits intoopensearch-project:mainfrom
JonahCalvo:data-loss-bug
Jul 30, 2025
Merged

Introduce checkpointing per shard to DDB source, via ShardAcknowledgementManager class#5818
graytaylor0 merged 14 commits intoopensearch-project:mainfrom
JonahCalvo:data-loss-bug

Conversation

@JonahCalvo
Copy link
Copy Markdown
Contributor

@JonahCalvo JonahCalvo commented Jun 25, 2025

Description

Introduce checkpointing per shard to DDB source, via ShardAcknowledgementManager class

Issues Resolved

Resolves #4764

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.

graytaylor0
graytaylor0 reviewed Jun 30, 2025
View reviewed changes
@graytaylor0
Copy link
Copy Markdown
Member

graytaylor0 commented Jun 30, 2025

DCO is failing as well.

dlvenable
dlvenable reviewed Jun 30, 2025
View reviewed changes
@JonahCalvo JonahCalvo force-pushed the data-loss-bug branch 2 times, most recently from efd7991 to 7f0ae3b Compare July 17, 2025 23:51
graytaylor0
graytaylor0 reviewed Jul 18, 2025
View reviewed changes
graytaylor0
graytaylor0 reviewed Jul 21, 2025
View reviewed changes
...a/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardAcknowledgementManager.java Outdated
}

public void giveUpPartition(final StreamPartition streamPartition) {
sourceCoordinator.saveProgressStateForPartition(streamPartition, dynamoDBSourceConfig.getShardAcknowledgmentTimeout());
Copy link
Copy Markdown
Member

@graytaylor0 graytaylor0 Jul 21, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the best approach here to basically just treat this signal as a failure, and handle it the same way you would handle a failure (i.e. do some checkpointing, stop tracking the partition for the next loop, and then give it up.

dlvenable
dlvenable requested changes Jul 23, 2025
View reviewed changes
graytaylor0
graytaylor0 reviewed Jul 25, 2025
View reviewed changes
...a/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardAcknowledgementManager.java
return globalPartition.isPresent();
}

public void startUpdatingOwnershipForShard(final StreamPartition streamPartition) {
Copy link
Copy Markdown
Member

@graytaylor0 graytaylor0 Jul 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would add a unit test where we make this call to ShardAcknowledgmentManager, and then run the loop and verify ownership update is called for that partition

Jonah Calvo added 5 commits July 30, 2025 11:30
Implement checkpointing of dynamodb stream source
2fb502f 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Fix unit tests
f73ecbc 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Do acknowledgements in batches per shard
28b31ac 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Minor changes to respond to PR feedback
8380392 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Better handling of failures during DDB acknowledgements
9980e80 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Jonah Calvo added 8 commits July 30, 2025 11:30
Update tracking of partitions to give up
ea8e1b5 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Add license headers
f853006 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Fix edge case where no records are received after initialization
8fe06ad 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Updated unit tests
ccbe7f8 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Only start updating ownership if Ack manager exists
4a4e007 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Fix shutdown logic in ack manager
7c858ac 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Minor changes to unit tests
528dfae 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Minor change to fix build issue
d7f17e4 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
revert change to build.gradle
860e78b 
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
@JonahCalvo JonahCalvo requested a review from dlvenable July 30, 2025 16:45
graytaylor0
graytaylor0 approved these changes Jul 30, 2025
View reviewed changes
...e/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumer.java
shardProgress.increment();
AcknowledgementSet acknowledgementSet = null;
if (shardAcknowledgementManager != null) {
acknowledgementSet = shardAcknowledgementManager.createAcknowledgmentSet(streamPartition, sequenceNumber, shardIterator == null);
Copy link
Copy Markdown
Member

@graytaylor0 graytaylor0 Jul 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know this is harder to verify but let's just keep an eye that shardIterator is actually null only once we reach the end of a s shard.

@graytaylor0 graytaylor0 merged commit dfc3c70 into opensearch-project:main Jul 30, 2025
54 of 59 checks passed
@dlvenable dlvenable mentioned this pull request Aug 1, 2025
Closed
@JonahCalvo JonahCalvo mentioned this pull request Aug 4, 2025
Merged
4 tasks
graytaylor0 added a commit to graytaylor0/data-prepper that referenced this pull request Aug 5, 2025
@graytaylor0
Revert "Introduce checkpointing per shard to DDB source, via ShardAck…
47846ef 
…nowledgementManager class (opensearch-project#5818)"

This reverts commit dfc3c70.

Signed-off-by: Taylor Gray <tylgry@amazon.com>
graytaylor0 added a commit that referenced this pull request Aug 5, 2025
@graytaylor0
Revert DDB checkpoint acknowledgment changes temporarily (#5967)
d040181 
* Revert "Ensure shards are completed when last getRecords call has no records and no shardIterator (#5958)"

This reverts commit 13049d3.

Signed-off-by: Taylor Gray <tylgry@amazon.com>

* Revert "Introduce checkpointing per shard to DDB source, via ShardAcknowledgementManager class (#5818)"

This reverts commit dfc3c70.

Signed-off-by: Taylor Gray <tylgry@amazon.com>

---------

Signed-off-by: Taylor Gray <tylgry@amazon.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@dlvenable dlvenable dlvenable approved these changes

@graytaylor0 graytaylor0 graytaylor0 approved these changes

@sb2k16 sb2k16 Awaiting requested review from sb2k16 sb2k16 is a code owner

@chenqi0805 chenqi0805 Awaiting requested review from chenqi0805

@engechas engechas Awaiting requested review from engechas engechas is a code owner

@san81 san81 Awaiting requested review from san81 san81 is a code owner

@srikanthjg srikanthjg Awaiting requested review from srikanthjg srikanthjg is a code owner

@dinujoh dinujoh Awaiting requested review from dinujoh dinujoh is a code owner

@kkondaka kkondaka Awaiting requested review from kkondaka kkondaka is a code owner

@KarstenSchnitter KarstenSchnitter Awaiting requested review from KarstenSchnitter KarstenSchnitter is a code owner

@oeyh oeyh Awaiting requested review from oeyh oeyh is a code owner

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

Checkpoint acknowledgments for DynamoDB pipelines

3 participants

@JonahCalvo @graytaylor0 @dlvenable