Add top-level filters option for S3 source by lawofcycles · Pull Request #6735 · opensearch-project/data-prepper

lawofcycles · 2026-04-06T21:06:02Z

Add a top-level filters configuration to the S3 source that applies include_prefix and exclude_suffix filtering for both SQS and scan modes.

Previously, key path filters were only available under scan bucket options, making it impossible to filter S3 objects when using SQS notifications. The new filters option uses the same bucket name keyed Map pattern as bucket_owners.

source:
 s3:
   filters:
     my-bucket:
       include_prefix:
          - assets/
       exclude_suffix:
          - .jpg
          - .xml

Top-level filters and scan bucket-level filter cannot be used together. A validation error is raised if both are configured.

Testing

In addition to unit tests, I verified this with a real S3 bucket and SQS queue (S3 ObjectCreated notifications forwarded to SQS). With include_prefix: ["assets/"] and exclude_suffix: [".jpg"] configured, the following results were observed.

S3 object key	Expected	Actual
`assets/data1.json`	Processed	Processed (read from S3, output to stdout)
`assets/photo.jpg`	Filtered out (exclude_suffix)	Filtered out (SQS message deleted)
`logs/app.log`	Filtered out (include_prefix mismatch)	Filtered out (SQS message deleted)
`root-file.json`	Filtered out (include_prefix mismatch)	Filtered out (SQS message deleted)

Issues Resolved

Resolves #6386

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 · 2026-04-06T21:06:30Z

✅ License Header Check Passed

All newly added files have proper license headers. Great work! 🎉

Add a top-level filters configuration to the S3 source that applies include_prefix and exclude_suffix filtering for both SQS and scan modes. Previously, key path filters were only available under scan bucket options, making it impossible to filter S3 objects when using SQS notifications. The new filters option uses the same bucket name keyed Map pattern as bucket_owners. Top-level filters and scan bucket-level filters cannot be used together, as the top-level filters are intended to eventually replace the scan bucket-level filters. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

dlvenable

Thank you @lawofcycles for this contribution! This will be a great help to users of SQS/EventBridge.

dlvenable · 2026-04-08T18:10:46Z

+/**
+ * Helper class for applying top-level S3 object key filters (include_prefix and exclude_suffix).
+ */
+public class S3ObjectFilteringHelper {


Thank you for splitting this logic into a class. Maybe rename it to S3ObjectKeyFilter or S3ObjectFilter.

Renamed to S3ObjectKeyFilter. Thanks for the suggestion!

dlvenable · 2026-04-08T18:13:23Z

+        when(keyPathOption.getS3ScanExcludeSuffixOptions()).thenReturn(null);
+        final S3ObjectFilteringHelper helper = new S3ObjectFilteringHelper(Map.of("my-bucket", keyPathOption));
+
+        assertThat(helper.isKeyMatchingFilters("my-bucket", "logs/app.log"), equalTo(false));


This could be combined with the test above using @ParameterizedTest and @CsvSource

@CsvSource({ "assets/image.png, true" "logs/app.log, false" })

Done, combined with @ParameterizedTest and @CsvSource.

dlvenable · 2026-04-08T18:13:38Z

+        when(keyPathOption.getS3ScanExcludeSuffixOptions()).thenReturn(List.of(".jpg", ".xml"));
+        final S3ObjectFilteringHelper helper = new S3ObjectFilteringHelper(Map.of("my-bucket", keyPathOption));
+
+        assertThat(helper.isKeyMatchingFilters("my-bucket", "assets/data.json"), equalTo(true));


Similar comment about combining tests.

Done as well.

dlvenable · 2026-04-08T18:14:03Z

+        when(keyPathOption.getS3ScanExcludeSuffixOptions()).thenReturn(List.of(".jpg"));
+        final S3ObjectFilteringHelper helper = new S3ObjectFilteringHelper(Map.of("my-bucket", keyPathOption));
+
+        assertThat(helper.isKeyMatchingFilters("my-bucket", "assets/data.json"), equalTo(true));


This is a good scenario for @CsvSource as well.

Done as well.

dlvenable · 2026-04-08T18:14:42Z

        return true;
    }

+    @AssertTrue(message = "Top-level filters cannot be used together with scan bucket-level filter. Use one or the other.")


Thank you for this assertion. It is the right approach in my opinion.

May I ask why we don't want to keep filters at both levels. For the case below that includes difference filters at two bucket level, it couldn't support SQS as the notification with a top level filter.

buckets: - bucket: name: offlinebatch data_selection: metadata_only filter: include_prefix: - bedrock-multisource/my_batch exclude_suffix: - .out - bucket: name: offlinebatch data_selection: data_only filter: include_prefix: - bedrock-multisource/output/ exclude_suffix: - manifest.json.out

@Zhangxunmt , The top-level filters are still able to differentiate buckets even with SQS. SQS includes the bucket name in the message.

Zhangxunmt · 2026-04-10T20:59:20Z

    }

+    @AssertTrue(message = "Top-level filters cannot be used together with scan bucket-level filter. Use one or the other.")
+    boolean isFiltersNotUsedWithScanBucketFilter() {


nit: Early returns (guard clauses) to reduce nesting for better readability.

boolean isFiltersNotUsedWithScanBucketFilter() { if (filters == null || filters.isEmpty()) return true; // top filter is null if (s3ScanScanOptions == null || s3ScanScanOptions.getBuckets() == null) return true; // bucket scan is null return s3ScanScanOptions.getBuckets().stream() .map(bucket -> bucket.getS3ScanBucketOption()) .noneMatch(option -> option != null && option.getS3ScanFilter() != null); // bucket filter is null }

Refactored to use guard clauses. Thanks!

Rename S3ObjectFilteringHelper to S3ObjectKeyFilter for clarity. Combine related test cases using @ParameterizedTest and @CsvSource. Refactor isFiltersNotUsedWithScanBucketFilter to use guard clauses. Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

lawofcycles force-pushed the feature/s3-source-top-level-filters branch from 6318b7e to 5e5fc0c Compare April 6, 2026 21:08

lawofcycles requested review from KarstenSchnitter, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners April 6, 2026 21:08

dlvenable requested changes Apr 8, 2026

View reviewed changes

Zhangxunmt reviewed Apr 10, 2026

View reviewed changes

lawofcycles requested a review from dlvenable April 18, 2026 16:38

dlvenable approved these changes Apr 21, 2026

View reviewed changes

Zhangxunmt approved these changes Apr 21, 2026

View reviewed changes

dlvenable merged commit 59d45a6 into opensearch-project:main Apr 21, 2026
69 of 72 checks passed

Conversation

lawofcycles commented Apr 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Testing

Issues Resolved

Check List

Uh oh!

github-actions Bot commented Apr 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ License Header Check Passed

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!

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 Apr 10, 2026 • 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!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

lawofcycles commented Apr 6, 2026 •

edited

Loading

github-actions Bot commented Apr 6, 2026 •

edited

Loading

Zhangxunmt Apr 10, 2026 •

edited

Loading