Implement LeaderOnlyTokenCrawler #6160

alparish · 2025-10-13T04:11:33Z

Description

This PR introduces the LeaderOnlyTokenCrawler, a performance-optimized implementation that eliminates redundant API calls by processing complete event content in the leader thread without worker partitions.

Created new LeaderOnlyTokenCrawler class that handles both content retrieval and buffer writing
Added support for both acknowledged and unacknowledged buffer writing
Implemented periodic checkpoint updates and failure handling

Issues Resolved

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

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.

alparish · 2025-10-13T19:52:11Z

.../opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawlerClient.java

+     *
+     * @param buffer The buffer to write events to
+     */
+    void setBuffer(Buffer<Record<Event>> buffer);


I will remove this method (setBuffer) and add buffer as a parameter to writeBatchToBuffer in a new revision

bbenner7635 · 2025-10-13T19:34:40Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+        long startTime = System.currentTimeMillis();
+        Instant lastCheckpointTime = Instant.now();
+
+        try {


Why do we need this try-catch?

Yes, the outer try catch is not needed as we already have a try catch around process batch. I will remove this.

bbenner7635 · 2025-10-13T22:35:41Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+
+            Iterator<ItemInfo> itemIterator = client.listItems(lastToken);
+
+            while (itemIterator.hasNext()) {


How come we switched from do-while to while loop? Ref

If itemIterator.hasNext() is false, then client.listItems() returned an empty iterator and there are no items to process. So there's no need to process anything if hasNext() is false on the first check. That's why a regular while loop is appropriate here.

bbenner7635 · 2025-10-13T22:54:59Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+            throw new RuntimeException("Crawl operation failed", e);
+        }
+
+        log.info("Crawl completed in {} ms", System.currentTimeMillis() - startTime);


Should we be recording crawl metric time same as other crawler classes?
https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/saas-source-plugins/source-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/source_crawler/base/TokenPaginationCrawler.java#L90

Included crawl metric

bbenner7635 · 2025-10-13T22:56:07Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+    private final LeaderOnlyTokenCrawlerClient client;
+    private final PluginMetrics pluginMetrics;
+    @Setter
+    private boolean acknowledgementsEnabled;


Is this being set from the Source plugin?

bbenner7635 · 2025-10-13T22:57:23Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+    private void processBatch(List<ItemInfo> batch,
+                              LeaderPartition leaderPartition,
+                              EnhancedSourceCoordinator coordinator) {
+        if (acknowledgementsEnabled && acknowledgementSetManager != null) {


If we expect acknowledgementSetManager to always exist if acknowledgementsEnabled is true, does it make sense to remove this null check? I am referring to WorkerScheduler

Open to your feedback.

Yes, you're right. Since acknowledgementSetManager is required when acknowledgements are enabled, the null check is redundant. I will remove it.

bbenner7635 · 2025-10-13T23:03:48Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+                        } else {
+                            // On failure: give up partition
+                            log.error("Batch processing failed for token: {}", lastToken);
+                            coordinator.giveUpPartition(leaderPartition);


I'm not sure if we should be giving up the partition, since this indicates the partition is shutting down and if this is for leader partition, then it will indicate the entire plugin is shutting down which is not the case:

Should be called by the source when it is shutting down to indicate that it will no longer be able to perform work on partitions

I followed documentdb's implementation, which also gives up the partition on negative acknowledgment https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/stream/StreamAcknowledgementManager.java#L103

@alparish what happens if acknowledgements do not comeback? Do you have a retry mechanism?

I agree with @bbenner7635 Giving up is probably not right here. Please test it locally to confirm the behavior

Ideally, acknowledgement timeouts should be very large (or INT MAX) because retrying events (that have not been acknowledged) means that the duplicate events will be sent. And if there are buffer issues or sink issues that caused the acknowledgements to be timedout, then injecting same events again will make the problem worse.

bbenner7635 · 2025-10-13T23:04:19Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+            AcknowledgementSet acknowledgementSet = acknowledgementSetManager.create(
+                    success -> {
+                        if (success) {
+                            // On success: update checkpoint


Do we need these plugin metrics?

Added these metrics

wjyao0316 · 2025-10-14T18:32:10Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+                    success -> {
+                        if (success) {
+                            // On success: update checkpoint
+                            updateLeaderProgressState(leaderPartition, lastToken, coordinator);


String a not a primitive variable. updateLeaderProgressState here is async function which might be invoked a while later.

Is it possible the lastToken here is already changed to a different value?

The callback is processed right after writing to the buffer, and before processing the next batch. Since we only update lastToken when processing a new batch, its value will remain consistent during the acknowledgment handling for the current batch.

I see. Then it could be a big performance risk because it will wait until event being ingested and then move to next batch.

Say the buffer wait time is 30 second, it means we can only process 50 events every 30 second?

If that is the case, shall we increase the batch size?

If we need to increase the batch size, we need a reasonably high number to ensure the minimal speed as well as not breaching buffer size.

@san81 Do you have any recommendation for the batch size

It is hard to just say one number. I would say, really test with different batch sizes and see what gives the best performance. The factors to consider are

vendor api support for max page size and latency with respect to the increase in page size

size of the api response payload with respect to the increase in page size. Just giving an example here. For S3 based pipelines, OSI pipelines process at least 20Mbps per OCU.

In these 3p connector pipelines, I don't think we will ever fill up the buffer. Each OCU comes up with 8GB buffer size. Biggest bottleneck is the vendor API latency and network latency.

If that is the case, I think we can easily go up to 5000 event batch for now and we will do load test to verify it is working without problem. The max size is no more then 10kb so the total size is maximumly 50MB, way less than 8GB.

After I rethink about it. We should make it configurable per connector.

For Okta, each API can only return maximum 100 events. So 5000 event batch requires 50 calls, which can easily run into a failure mode where 1 api failure block 49 other calls.
A reasonable balance is 5 ~ 10 calls

Signed-off-by: Alekhya Parisha <[email protected]>

san81 · 2025-10-16T22:49:50Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+                processBatch(batch, leaderPartition, coordinator);
+            } catch (Exception e) {
+                batchesFailedCounter.increment();
+                log.error("Failed to process batch ending with token {}", lastToken, e);


Using NOISY is preferred for this case since we are printing the entire stack trace

I'm not sure about this. This is happening at a batch level (not per-event level). And it should be more of a defensive programming catch.

san81 · 2025-10-16T23:06:03Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+                        } else {
+                            // On failure: give up partition
+                            log.error("Batch processing failed for token: {}", lastToken);
+                            coordinator.giveUpPartition(leaderPartition);


I agree with @bbenner7635 Giving up is probably not right here. Please test it locally to confirm the behavior

dlvenable

Something about this seems to mix concerns. We already have four crawlers. Each one has a different way to crawl. This change is still the token crawler, but it is using only the leader node. We should decouple the approach for crawling from the nodes that make use of it. Otherwise, we may end up with 8 crawlers.

dlvenable · 2025-10-17T17:51:31Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+                processBatch(batch, leaderPartition, coordinator);
+            } catch (Exception e) {
+                batchesFailedCounter.increment();
+                log.error("Failed to process batch ending with token {}", lastToken, e);


I'm not sure about this. This is happening at a batch level (not per-event level). And it should be more of a defensive programming catch.

dlvenable · 2025-10-17T17:53:00Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+@Named
+public class LeaderOnlyTokenCrawler implements Crawler {
+    private static final Logger log = LoggerFactory.getLogger(LeaderOnlyTokenCrawler.class);
+    private static final Duration BUFFER_WRITE_TIMEOUT = Duration.ofSeconds(15);


Shouldn't the client be providing these values?

dlvenable · 2025-10-17T17:58:09Z

...va/org/opensearch/dataprepper/plugins/source/source_crawler/base/LeaderOnlyTokenCrawler.java

+import java.util.concurrent.TimeUnit;
+
+@Named
+public class LeaderOnlyTokenCrawler implements Crawler {


This should not use the raw type. You need Crawler<...>.

alparish requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners October 13, 2025 04:11

alparish commented Oct 13, 2025

View reviewed changes

bbenner7635 reviewed Oct 13, 2025

View reviewed changes

wjyao0316 reviewed Oct 14, 2025

View reviewed changes

alparish force-pushed the feature/leaderOnlyTokenCrawler branch from f9b45fc to c2fd947 Compare October 15, 2025 03:53

Implement LeaderOnlyTokenCrawler

ca88f1c

Signed-off-by: Alekhya Parisha <[email protected]>

alparish force-pushed the feature/leaderOnlyTokenCrawler branch from c2fd947 to ca88f1c Compare October 15, 2025 03:57

bbenner7635 approved these changes Oct 15, 2025

View reviewed changes

san81 reviewed Oct 16, 2025

View reviewed changes

dlvenable reviewed Oct 17, 2025

View reviewed changes


		Iterator<ItemInfo> itemIterator = client.listItems(lastToken);

		while (itemIterator.hasNext()) {

Implement LeaderOnlyTokenCrawler #6160

Are you sure you want to change the base?

Implement LeaderOnlyTokenCrawler #6160

Uh oh!

Conversation

alparish commented Oct 13, 2025

Description

Issues Resolved

Check List

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kkondaka Oct 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

kkondaka Oct 16, 2025 •

edited

Loading