[graytaylor0]

… folder partitioning

Description

Changes

Adds a new idle_scan_mode configuration to S3 folder partitions with two modes:

idle_optimized (default) — Current behavior. Empty folder partitions are given up with MAX_RETRIES, allowing TTL expiry. Minimizes S3 ListObjectsV2 API costs when idle.
continuous — Empty folder partitions are given up with 0 retries for immediate re-acquisition. Workers keep scanning all folders with the standard 30s backoff, achieving sub-minute idle detection at the cost of additional S3 LIST calls (~$36/month per idle OCU).

The change spans three layers:

s3-source plugin — New IdleScanMode enum, idle_scan_mode field in FolderPartitioningOptions, and mode-aware empty-folder handling in ScanObjectWorker.processFolderPartition().
mongodb plugin — New s3_scan_mode field in CollectionConfig and MongoDBSourceConfig (defaults to idle_optimized).
documentdb template — Propagates s3_scan_mode from the DocumentDB source config to idle_scan_mode in the generated S3 source folder_partitions config.

Usage
Copy

source:
documentdb:
s3_scan_mode: "continuous"
collections:
- collection: "mydb.mycollection"
export: true
stream: true
...

Issues Resolved

Resolves #7170
Resolves #6051

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.

[github-actions]

⚠️ License Header Violations Found

The following newly added files are missing required license headers:

• data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/configuration/IdleScanMode.java

Please add the appropriate license header to each file and push your changes.

See the license header requirements: https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md#license-headers

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Add idle_scan_mode for S3 folder partitions and plumb through DocumentDB Relevant files: data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ScanObjectWorker.java data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/configuration/FolderPartitioningOptions.java data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/configuration/IdleScanMode.java data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/S3ScanObjectWorkerTest.java data-prepper-plugins/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/configuration/CollectionConfig.java data-prepper-plugins/mongodb/src/main/java/org/opensearch/dataprepper/plugins/mongo/configuration/MongoDBSourceConfig.java data-prepper-plugins/mongodb/src/main/resources/org/opensearch/dataprepper/transforms/templates/documentdb-template.yaml Sub-PR theme: Add sink thread-pool queue wait time metric Relevant files: data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/AbstractSink.java data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/Sink.java data-prepper-core/src/main/java/org/opensearch/dataprepper/core/pipeline/Pipeline.java Sub-PR theme: Add OpenSearch bulk HTTP/serialization latency metrics Relevant files: data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

⚡ Recommended focus areas for review Possible Issue In continuous scan mode, when the folder partition is empty the code calls giveUpPartition and returns early, but it never clears/completes the state transitions that the subsequent block does (e.g. shouldDeleteFolderPartition path). More importantly, giving up with 0 retries causes the same worker (or peers) to immediately re-acquire the same empty partition in a tight loop, since the standard 30s backoff mentioned in the PR description is per-worker scheduling, not per-partition. Verify that there is a per-partition cooldown so this does not create a hot loop generating constant S3 LIST calls (and log spam at DEBUG) when all folders are idle. if (isContinuousScanMode()) { LOG.debug("No objects to process in continuous scan mode, giving up partition for immediate re-acquisition"); sourceCoordinator.giveUpPartition(folderPartition.getPartitionKey(), Instant.now(), 0); return; } Validation Missing s3ScanMode is accepted as a free-form String with no validation (no @Pattern or enum mapping). A typo such as "Continuous" or "idleoptimized" will be silently propagated into the generated S3 pipeline where IdleScanMode.fromString will throw IllegalArgumentException at pipeline startup, far from the misconfigured field. Consider mapping to the IdleScanMode enum directly (or validating with a pattern) so bad values are rejected with a clear error at config load. @JsonProperty("s3_scan_mode") private String s3ScanMode; Inconsistent Default/Scope MongoDBSourceConfig.s3ScanMode and CollectionConfig.s3ScanMode both exist and both default to "idle_optimized", but only the source-level value is referenced from the documentdb template ($.source.documentdb.s3_scan_mode). The per-collection s3_scan_mode field is therefore accepted in YAML but has no effect, which will confuse users. Either remove the per-collection field or wire it through the template. @JsonProperty("s3_scan_mode") private String s3ScanMode = "idle_optimized"; Possible NPE fromString calls name.toLowerCase() without a null check. If Jackson invokes the creator with a null (e.g. when the template substitution <<$...s3_scan_mode>> resolves to an absent/null value), this throws NullPointerException instead of a configuration error or falling back to the default. Consider returning the default (or throwing a clear message) when name == null. public static IdleScanMode fromString(final String name) { final IdleScanMode mode = NAMES_MAP.get(name.toLowerCase()); if (mode == null) { throw new IllegalArgumentException("Invalid idle_scan_mode: " + name + ". Valid values are: " + NAMES_MAP.keySet()); } return mode; } Incorrect Metric bulkRequestSerializationTimer records the elapsed time between serializationStart and the next line that only calls bulkRequest.getRequest() — no serialization occurs there. The timer will always report near-zero and does not measure what its name implies. Either remove this timer or move the measurement around the actual serialization call site. bulkRequest -> { final long serializationStart = System.nanoTime(); final org.opensearch.client.opensearch.core.BulkRequest request = bulkRequest.getRequest(); bulkRequestSerializationTimer.record(System.nanoTime() - serializationStart, TimeUnit.NANOSECONDS); final long httpStart = System.nanoTime(); try { return bulkApiWrapper.bulk(request); } finally { bulkRequestHttpTimer.record(System.nanoTime() - httpStart, TimeUnit.NANOSECONDS); } },

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix misleading serialization timer measurement The bulkRequestSerializationTimer currently measures only the bulkRequest.getRequest() call, which is almost certainly a trivial getter — not actual serialization. The real serialization happens inside bulkApiWrapper.bulk(request) and is now counted in the HTTP timer, making both metrics misleading. Either remove the serialization timer or instrument it where serialization actually occurs. data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java [236-246] bulkRequest -> { - final long serializationStart = System.nanoTime(); final org.opensearch.client.opensearch.core.BulkRequest request = bulkRequest.getRequest(); - bulkRequestSerializationTimer.record(System.nanoTime() - serializationStart, TimeUnit.NANOSECONDS); final long httpStart = System.nanoTime(); try { return bulkApiWrapper.bulk(request); } finally { bulkRequestHttpTimer.record(System.nanoTime() - httpStart, TimeUnit.NANOSECONDS); } }, Suggestion importance[1-10]: 7 __ Why: Valid observation: the bulkRequestSerializationTimer wraps only bulkRequest.getRequest(), which is a simple getter, making the serialization metric misleading. This is a legitimate code quality/correctness concern.	Medium
General	Provide default for optional template parameter If s3_scan_mode is not provided in the DocumentDB source configuration, this template substitution may resolve to null/empty and cause IdleScanMode.fromString to fail. Ensure the transform supplies a default (e.g., idle_optimized) when the user omits it, matching MongoDBSourceConfig's default. data-prepper-plugins/mongodb/src/main/resources/org/opensearch/dataprepper/transforms/templates/documentdb-template.yaml [73] folder_partitions: depth: "<<FUNCTION_NAME:calculateDepth,PARAMETER:$.<<pipeline-name>>.source.documentdb.s3_prefix>>" max_objects_per_ownership: 50 - idle_scan_mode: "<<$.<<pipeline-name>>.source.documentdb.s3_scan_mode>>" + idle_scan_mode: "<<$.<<pipeline-name>>.source.documentdb.s3_scan_mode:idle_optimized>>" Suggestion importance[1-10]: 7 __ Why: Important issue: if users don't supply s3_scan_mode, the template substitution may produce an invalid value causing pipeline initialization to fail. Providing a default matches the config class defaults.	Medium
	Avoid tight re-acquisition loop on empty folders In continuous scan mode, giving up the partition immediately with zero priority can cause a tight re-acquisition loop that hammers S3 ListObjects and the coordinator when a folder stays empty. Consider adding a minimum back-off (e.g., using a non-zero reacquire delay) to prevent CPU/API pressure while still enabling frequent polling. data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/ScanObjectWorker.java [348-355] if (objectsToProcess.isEmpty()) { folderPartitionNoObjectsFound.increment(); partitionKeys.remove(folderPartition.getPartitionKey()); if (isContinuousScanMode()) { - LOG.debug("No objects to process in continuous scan mode, giving up partition for immediate re-acquisition"); - sourceCoordinator.giveUpPartition(folderPartition.getPartitionKey(), Instant.now(), 0); + LOG.debug("No objects to process in continuous scan mode, giving up partition for re-acquisition"); + sourceCoordinator.giveUpPartition(folderPartition.getPartitionKey(), Instant.now().plusSeconds(1), 0); return; } Suggestion importance[1-10]: 6 __ Why: Reasonable concern about tight re-acquisition causing S3 API pressure, though the "continuous" mode name implies intentional immediate re-polling. The suggestion is sensible but could conflict with design intent.	Low
	Guard enum factory against null input name.toLowerCase() will throw a NullPointerException if a null value is supplied (e.g., unresolved template placeholder). Guard against null input and throw a clearer IllegalArgumentException to aid configuration debugging. data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/configuration/IdleScanMode.java [32-39] @JsonCreator public static IdleScanMode fromString(final String name) { + if (name == null) { + throw new IllegalArgumentException("idle_scan_mode must not be null. Valid values are: " + NAMES_MAP.keySet()); + } final IdleScanMode mode = NAMES_MAP.get(name.toLowerCase()); if (mode == null) { throw new IllegalArgumentException("Invalid idle_scan_mode: " + name + ". Valid values are: " + NAMES_MAP.keySet()); } return mode; } Suggestion importance[1-10]: 5 __ Why: Valid defensive coding suggestion, especially relevant given the template-driven config could pass unresolved values. However, it's a minor robustness improvement.	Low