Add S3 Scan processing condition evaluator to ensure object completeness by Zhangxunmt · Pull Request #6624 · opensearch-project/data-prepper

Zhangxunmt · 2026-03-10T06:57:25Z

Description

Problem

When using S3 scan to read the output of asynchronous batch inference jobs (e.g. Amazon Bedrock batch jobs), output files appear in S3 before the job has fully completed. There was no way to delay processing of those files until the job was actually done, leading to partial reads.

Solution

Introduces a new processing_conditions list at the S3 scan bucket level. Before processing an S3 object, the worker evaluates each applicable condition by reading a co-located manifest/sentinel file and running a Data Prepper expression against its JSON content. Processing is deferred until all conditions are satisfied.

Configuration example

  scan:
    buckets:
      - bucket:
          name: "my-batch-bucket"
          data_selection: data_only
          processing_conditions:
            - object_name: "manifest.json.out"
              when: "/processedRecordCount == /totalRecordCount"
              applicable_prefix:
                - "bedrock-multisource/output-multisource/"
              retry_delay: PT5M
              max_retry: 10
          filter:
            include_prefix:
              - "bedrock-multisource/output-multisource/"
            exclude_suffix:
              - manifest.json.out

For an object at s3://my-bucket/bedrock-multisource/output-multisource/job-123/result.jsonl.out, the worker will:

Download s3://my-bucket/bedrock-multisource/output-multisource/job-123/manifest.json.out
Parse it as JSON and evaluate /processedRecordCount == /totalRecordCount
If true → proceed with processing
If false or manifest not found → release the partition back with a now + retry_delay priority timestamp, retrying up to max_retry times

processing_conditions fields:

object_name │ string │ required │ Manifest file name, resolved in the same S3 directory as the object
when │ string │ required │ Data Prepper expression evaluated against manifest JSON
applicable_prefix │ list │ all objects │ Prefixes this condition applies to; absent means all objects in the bucket
retry_delay │ Duration │ PT5M │ Wait time before re-queuing when condition is not met
max_retry │ int │ 10 │ Max retry attempts before the partition is abandoned

Implementation details:

New classes:

S3ScanProcessingCondition — config POJO for a single condition entry
S3ScanProcessingConditionEvaluator — encapsulates all condition-checking logic: S3 manifest download, JSON parsing into JacksonEvent, expression evaluation via ExpressionEvaluator

Modified classes:

S3ScanBucketOption — adds processing_conditions field
ScanObjectWorker — checks conditions before processing each object; on failure calls sourceCoordinator.closePartition(objectToProcess.get().getPartitionKey(), retryDelay, maxRetry, false);
S3ScanService — constructs a single shared S3ScanProcessingConditionEvaluator (stateless) and passes it to all workers
S3Source — injects ExpressionEvaluator via @DataPrepperPluginConstructor and threads it through to S3ScanService

Testing:

23 unit tests added for S3ScanProcessingConditionEvaluator covering:
- Null/empty condition guards
- include_prefix matching (no match skips, empty/null applies to all, partial match)
- Expression evaluation (true/false, JSON fields forwarded correctly to evaluator)
- Manifest key resolution (nested path uses same directory; root-level object uses filename only)
- Error handling (NoSuchKeyException → defer; unexpected exception → defer)
- Multi-condition logic (all pass, first fails short-circuits, non-matching skipped)
- findFirstMatching for retry setting lookup
Existing unit and integration tests updated to compile with the new constructor signatures

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.

graytaylor0

Thanks for making this change. It looks good! Couple comments on what the behavior is for some of the edge cases.

graytaylor0 · 2026-03-13T17:58:22Z

+            }
+            final String manifestKey = resolveManifestKey(objectKey, condition.getFileName());
+            try {
+                final String content = readS3ObjectAsString(bucket, manifestKey);


How do you think we should handle the case where the manifest file is not json?

This is a good call out. I think we can add a codec section in the processing condition config. Thoughts?

Yes, I think a codec is ideal here.

Added codec to process different formats of the object for conditions. It'd be one codec per condition so I used a Map to load all the codec.

graytaylor0 · 2026-03-13T19:17:33Z

Also one more thing. As a follow up please create a follow up PR to the documentation website here (https://github.com/opensearch-project/documentation-website).

This documentation is what users will use (https://docs.opensearch.org/latest/data-prepper/pipelines/configuration/sources/s3/#scan-bucket)

dlvenable

Thanks @Zhangxunmt !

dlvenable · 2026-03-16T09:15:20Z

+            }
+            final String manifestKey = resolveManifestKey(objectKey, condition.getFileName());
+            try {
+                final String content = readS3ObjectAsString(bucket, manifestKey);


Yes, I think a codec is ideal here.

dlvenable · 2026-03-16T09:21:22Z

+    @NotEmpty
+    private String when;
+
+    @JsonProperty("include_prefix")


I think this name may be a little confusing. It initially appeared to me to relate to whether or not to include the condition file. But instead it means that the condition is applicable for any object in that prefix.

It seems we may need a slightly better name to help clarify this. Perhaps "applicable" would be clearer - applicable_for_prefix.

renamed to applicable_prefix

dlvenable · 2026-03-16T09:23:31Z


+        final List<S3ScanProcessingCondition> conditions = bucketProcessingConditionsMap.get(bucket);
+        if (!processingConditionEvaluator.allConditionsMet(bucket, objectKey, conditions)) {
+            final S3ScanProcessingCondition failedCondition = processingConditionEvaluator.findFirstMatching(objectKey, conditions);


This may not be the failed condition, right? It seems we are just arbitrarily picking the first one.

yeah probably not accurate enough. refactored to use the exact failed condition to proceed.

dlvenable · 2026-03-16T09:30:10Z

+            if (!isApplicable(objectKey, condition)) {
+                continue;
+            }
+            final String manifestKey = resolveManifestKey(objectKey, condition.getFileName());


Let's remove reference to "manifest" since that is a Bedrock term.

renamed to conditionObjectKey

…teness Signed-off-by: Xun Zhang <xunzh@amazon.com>

…hen max_retry exhauses etc Signed-off-by: Xun Zhang <xunzh@amazon.com>

graytaylor0 · 2026-03-17T04:20:58Z

+        final AtomicReference<Event> firstEvent = new AtomicReference<>();
+        final GetObjectRequest request = GetObjectRequest.builder().bucket(bucket).key(key).build();
+        try (final ResponseInputStream<GetObjectResponse> response = s3Client.getObject(request)) {
+            codec.parse(response, record -> firstEvent.compareAndSet(null, record.getData()));


One thing we should make sure of. Does this actually work with a single json object in a file? Last I checked the json codec only supported json arrays. Did you test again with a static json object (not ndjson)

Yes I tested and verified it works with single json object. I uses the actual one of the real manifest file in the below format.

{"totalRecordCount":50000,"processedRecordCount":50000,"successRecordCount":30000,"errorRecordCount":0,"inputTokenCount":5300000,"outputTokenCount":0,"inputAudioSecond":0,"inputVideoSecond":0,"inputStandardImageCount":0,"inputDocumentImageCount":0,"inputTextTokenCount":0,"inputImageTokenCount":0,"inputAudioTokenCount":0,"inputVideoTokenCount":0,"outputTextTokenCount":0,"outputImageTokenCount":0}

Zhangxunmt · 2026-03-17T22:53:48Z

Also one more thing. As a follow up please create a follow up PR to the documentation website here (https://github.com/opensearch-project/documentation-website).

This documentation is what users will use (https://docs.opensearch.org/latest/data-prepper/pipelines/configuration/sources/s3/#scan-bucket)

The document PR is created for this new config update: opensearch-project/documentation-website#12104.

dlvenable

Thanks @Zhangxunmt !

Zhangxunmt requested review from KarstenSchnitter, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners March 10, 2026 06:57

Zhangxunmt force-pushed the main branch from 932b73c to a6a1cc5 Compare March 12, 2026 23:55

graytaylor0 reviewed Mar 13, 2026

View reviewed changes

dlvenable requested changes Mar 16, 2026

View reviewed changes

add S3 Scan processing condition evaluator to ensure S3 object comple…

3885919

…teness Signed-off-by: Xun Zhang <xunzh@amazon.com>

Zhangxunmt force-pushed the main branch from a6a1cc5 to d358321 Compare March 17, 2026 00:23

address comments: add codec, rename config names, and process as is w…

7bbe3cf

…hen max_retry exhauses etc Signed-off-by: Xun Zhang <xunzh@amazon.com>

Zhangxunmt force-pushed the main branch from d358321 to 7bbe3cf Compare March 17, 2026 01:24

graytaylor0 reviewed Mar 17, 2026

View reviewed changes

graytaylor0 approved these changes Mar 18, 2026

View reviewed changes

oeyh approved these changes Mar 18, 2026

View reviewed changes

dlvenable approved these changes Mar 19, 2026

View reviewed changes

dlvenable merged commit 363af8f into opensearch-project:main Mar 19, 2026
72 checks passed

dlvenable mentioned this pull request Apr 29, 2026

Adds Xun Zhang (Zhangxunmt) to the maintainers. #6793

Merged

4 tasks

Conversation

Zhangxunmt commented Mar 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Issues Resolved

Check List

Uh oh!

graytaylor0 left a comment

Choose a reason for hiding this comment

Uh oh!

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!

graytaylor0 commented Mar 13, 2026

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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!

Zhangxunmt commented Mar 17, 2026

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Zhangxunmt commented Mar 10, 2026 •

edited

Loading