Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add AcknowledgementSet support to DocumentDB/MongoDB streams #4379

Merged
merged 18 commits into from
Apr 2, 2024
Merged

Add AcknowledgementSet support to DocumentDB/MongoDB streams #4379

merged 18 commits into from
Apr 2, 2024

Conversation

dinujoh
Copy link
Member

@dinujoh dinujoh commented Apr 1, 2024

Description

Add AcknowledgementSet support to DocumentDB/MongoDB streams

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.

@dinujoh
Add AcknowledgementSet support to DocumentDB/MongoDB streams
d16feb4 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Update StreamAcknowledgementManagerTest
9c7e6b8 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
chenqi0805
chenqi0805 reviewed Apr 1, 2024
View reviewed changes
.../main/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamAcknowledgementManager.java
}
}
parentThread.interrupt();
executorService.shutdown();
Copy link
Collaborator

Choose a reason for hiding this comment

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

Should L88-L89 be in init method instead of the callable?

Copy link
Member Author

Choose a reason for hiding this comment

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

No, once this monitor exists because of ack wait timeout, it should stop the stream worker.

@dinujoh
Unit test update
e7d3b94 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Update unit test
4def974 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Update unit test
5740266 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Unit test fix
a1417a7 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Unit test fix
85c9894 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Unit test fix
bdd789b 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Update unit test
d503063 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Update test
3444d1e 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Unit test fix
1b9749c 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Unit test fix
82369fc 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Fix unit test
f77c08a 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Unit test fix
bd4c251 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Disabling test
99a73d3 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
dlvenable
dlvenable requested changes Apr 1, 2024
View reviewed changes
Copy link
Member

@dlvenable dlvenable left a comment

Choose a reason for hiding this comment

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

I've only taken a cursory look and haven't looked at the logic yet.

.../main/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamAcknowledgementManager.java Outdated
this.partitionAcknowledgmentTimeout = partitionAcknowledgmentTimeout;
this.acknowledgementMonitorWaitTimeInMs = acknowledgementMonitorWaitTimeInMs;
this.checkPointIntervalInMs = checkPointIntervalInMs;
executorService = Executors.newSingleThreadExecutor();
Copy link
Member

Choose a reason for hiding this comment

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

Let's use the BackgroundThreadFactory for all ExecutorServices going forward.

Here is an example of using it:

data-prepper/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/SqsService.java

Line 48 in 19b18a1

executorService = Executors.newFixedThreadPool(s3SourceConfig.getNumWorkers(), BackgroundThreadFactory.defaultExecutorThreadFactory("s3-source-sqs"));

This thread factory will give us more useful names and also make the thread a daemon thread to ensure Data Prepper shuts down properly.

Copy link
Member Author

Choose a reason for hiding this comment

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

Done

...odb/src/main/java/org/opensearch/dataprepper/plugins/mongo/documentdb/DocumentDBService.java Outdated
executor.submit(exportScheduler);
executor.submit(exportWorker);
executor.submit(streamScheduler);
executor = Executors.newFixedThreadPool(runnableList.size());
Copy link
Member

Choose a reason for hiding this comment

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

Also, use the BackgroundThreadFactory here.

Copy link
Member Author

Choose a reason for hiding this comment

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

Done

...gins/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamWorker.java Outdated
if (recordCount % defaultFlushBatchSize == 0) {
LOG.debug("Write to buffer for line " + (recordCount - defaultFlushBatchSize) + " to " + recordCount);
if (recordCount % recordFlushBatchSize == 0) {
LOG.debug("Write to buffer for line " + (recordCount - recordFlushBatchSize) + " to " + recordCount);
Copy link
Member

Choose a reason for hiding this comment

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

Please use string interpolation:

LOG.debug("Write to buffer for line {} to {}", (recordCount - recordFlushBatchSize), recordCount);

Copy link
Member Author

Choose a reason for hiding this comment

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

Done

@dinujoh
Add stop method to Stream worker to stop processing stream
5d0c291 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Add BackgroundThreadFactory that adds thread name prefix for debugging
22b3685 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh
Update unit test
cf4b857 
Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>
@dinujoh dinujoh requested a review from dlvenable April 2, 2024 14:24
dlvenable
dlvenable requested changes Apr 2, 2024
View reviewed changes
...s/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/model/CheckpointStatus.java
@@ -0,0 +1,45 @@
package org.opensearch.dataprepper.plugins.mongo.model;

public class CheckpointStatus {
Copy link
Member

Choose a reason for hiding this comment

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

This class is only used by StreamAcknowledgementManager. So you can move this into the stream package and make it package-private to help keep it encapsulated.

Copy link
Member Author

Choose a reason for hiding this comment

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

Sure. will do.

.../main/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamAcknowledgementManager.java
}
LOG.warn("Acknowledgement not received for the checkpoint {} past wait time. Giving up partition.", checkpointStatus.getResumeToken());
partitionCheckpoint.giveUpPartition();
break;
Copy link
Member

Choose a reason for hiding this comment

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

I don't understand this. This will end the thread and then also the stream thread. But, would these ever start again? It seems that when this happens, this node is now done working and unable to contribute.

Copy link
Member Author

Choose a reason for hiding this comment

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

Yes, this will end the stream. New node or same node can acquire the partition and work from the checkpoint state.

Copy link
Member

Choose a reason for hiding this comment

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

OK. I see. The StreamScheduler is still running and can re-acquire.

@dinujoh dinujoh merged commit b7c63bc into opensearch-project:main Apr 2, 2024
46 of 47 checks passed
@kkondaka kkondaka added this to the v2.8 milestone May 14, 2024
@dinujoh
Copy link
Member Author

dinujoh commented May 14, 2024

#4534

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

@chenqi0805 chenqi0805 chenqi0805 approved these changes

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

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

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

@asifsmohammed asifsmohammed Awaiting requested review from asifsmohammed

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

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
v2.8
Development

Successfully merging this pull request may close these issues.
4 participants
@dinujoh @dlvenable @chenqi0805 @kkondaka