From de8966c7f41720b809d5849a46d69f3670d96a00 Mon Sep 17 00:00:00 2001 From: Aditya Kumar Singh Date: Thu, 22 Oct 2026 18:21:27 +0530 Subject: [PATCH 1/4] fix(s3): filter objects during pagination in S3Service.list to avoid premature loop exit --- .../io/kestra/plugin/aws/s3/S3Service.java | 23 ++--- .../plugin/aws/s3/S3ServiceListTest.java | 89 +++++++++++++++++++ 2 files changed, 98 insertions(+), 14 deletions(-) diff --git a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java index a193892a..1d4c7881 100644 --- a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java +++ b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java @@ -32,7 +32,6 @@ import software.amazon.awssdk.transfer.s3.model.DownloadFileRequest; import software.amazon.awssdk.transfer.s3.model.FileDownload; -import static io.kestra.core.utils.Rethrow.throwPredicate; public class S3Service { @@ -275,23 +274,24 @@ public static List list(RunContext runContext, S3Client client, ListIn var filter = runContext.render(list.getFilter()).as(ListInterface.Filter.class).orElseThrow(); ListObjectsV2Request baseRequest = builder.build(); - List allContents = new ArrayList<>(); + List result = new ArrayList<>(); ListObjectsV2Request currentRequest = baseRequest; ListObjectsV2Response response; do { response = client.listObjectsV2(currentRequest); - allContents.addAll(response.contents()); - - // maxKeys is a total cap across pages, the per-page size is capped at 1000 by S3. - if (allContents.size() >= rMaxKeys) { - break; + for (var s3Object : response.contents()) { + if (S3Service.filter(s3Object, regExp, filter)) { + result.add(S3Object.of(s3Object)); + if (result.size() >= rMaxKeys) { + return result; + } + } } if (response.isTruncated()) { String nextToken = response.nextContinuationToken(); if (nextToken == null) { - // Guard: isTruncated=true but no token returned, stop to avoid an infinite loop. runContext.logger().warn("S3 list pagination stopped: isTruncated=true but no continuation token available."); break; } @@ -299,12 +299,7 @@ public static List list(RunContext runContext, S3Client client, ListIn } } while (response.isTruncated()); - var contents = allContents.size() > rMaxKeys ? allContents.subList(0, rMaxKeys) : allContents; - - return contents.stream() - .filter(throwPredicate(s3Object -> S3Service.filter(s3Object, regExp, filter))) - .map(S3Object::of) - .toList(); + return result; } private static boolean filter(software.amazon.awssdk.services.s3.model.S3Object object, String regExp, ListInterface.Filter filter) { diff --git a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java index 4029a81c..ceaf5482 100644 --- a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java +++ b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java @@ -74,4 +74,93 @@ void list_paginatesAcrossTwoTruncatedPages() throws Exception { // Verify the second call carried the continuation token. verify(mockClient, times(2)).listObjectsV2(any(ListObjectsV2Request.class)); } + + @Test + void shouldContinuePaginationWhenInitialPageDoesNotMatchFilter() throws Exception { + RunContext runContext = runContextFactory.of(); + + var obj1 = software.amazon.awssdk.services.s3.model.S3Object.builder() + .key("archive/old1.txt").size(10L).build(); + var obj2 = software.amazon.awssdk.services.s3.model.S3Object.builder() + .key("archive/old2.txt").size(20L).build(); + var obj3 = software.amazon.awssdk.services.s3.model.S3Object.builder() + .key("data/orders.csv").size(30L).build(); + + var page1 = ListObjectsV2Response.builder() + .isTruncated(true) + .nextContinuationToken("token-page2") + .contents(obj1, obj2) + .build(); + + var page2 = ListObjectsV2Response.builder() + .isTruncated(false) + .contents(obj3) + .build(); + + S3Client mockClient = mock(S3Client.class); + when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) + .thenAnswer(inv -> { + ListObjectsV2Request req = inv.getArgument(0); + return "token-page2".equals(req.continuationToken()) ? page2 : page1; + }); + + var task = io.kestra.plugin.aws.s3.List.builder() + .id("s3-service-list-pagination-filter") + .type(io.kestra.plugin.aws.s3.List.class.getName()) + .bucket(Property.ofValue("my-bucket")) + .regexp(Property.ofValue(".*\\.csv")) + .maxKeys(Property.ofValue(2)) + .build(); + + List result = S3Service.list(runContext, mockClient, task, task); + + assertThat(result, hasSize(1)); + assertThat(result.get(0).getKey(), is("data/orders.csv")); + verify(mockClient, times(2)).listObjectsV2(any(ListObjectsV2Request.class)); + } + + @Test + void shouldStopPaginationWhenMatchingCountReachesMaxKeys() throws Exception { + RunContext runContext = runContextFactory.of(); + + var obj1 = software.amazon.awssdk.services.s3.model.S3Object.builder() + .key("orders-1.csv").size(10L).build(); + var obj2 = software.amazon.awssdk.services.s3.model.S3Object.builder() + .key("orders-2.csv").size(20L).build(); + var obj3 = software.amazon.awssdk.services.s3.model.S3Object.builder() + .key("orders-3.csv").size(30L).build(); + + var page1 = ListObjectsV2Response.builder() + .isTruncated(true) + .nextContinuationToken("token-page2") + .contents(obj1, obj2) + .build(); + + var page2 = ListObjectsV2Response.builder() + .isTruncated(false) + .contents(obj3) + .build(); + + S3Client mockClient = mock(S3Client.class); + when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) + .thenAnswer(inv -> { + ListObjectsV2Request req = inv.getArgument(0); + return "token-page2".equals(req.continuationToken()) ? page2 : page1; + }); + + var task = io.kestra.plugin.aws.s3.List.builder() + .id("s3-service-list-max-keys") + .type(io.kestra.plugin.aws.s3.List.class.getName()) + .bucket(Property.ofValue("my-bucket")) + .regexp(Property.ofValue(".*\\.csv")) + .maxKeys(Property.ofValue(2)) + .build(); + + List result = S3Service.list(runContext, mockClient, task, task); + + assertThat(result, hasSize(2)); + assertThat(result.get(0).getKey(), is("orders-1.csv")); + assertThat(result.get(1).getKey(), is("orders-2.csv")); + verify(mockClient, times(1)).listObjectsV2(any(ListObjectsV2Request.class)); + } } From 64c6c54efa42b3e6550e0bbf2f40257c16ca550b Mon Sep 17 00:00:00 2001 From: Malay Dewangan Date: Mon, 26 Oct 2026 04:00:17 +0530 Subject: [PATCH 2/4] fix(s3): request full pages when filtering, guard maxKeys 0 and compile the regExp once --- .../kestra/plugin/aws/s3/ListInterface.java | 2 +- .../io/kestra/plugin/aws/s3/S3Service.java | 22 ++++--- .../plugin/aws/s3/S3ServiceListTest.java | 58 +++++++++++++++---- 3 files changed, 62 insertions(+), 20 deletions(-) diff --git a/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java b/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java index c603310c..28c8a303 100644 --- a/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java +++ b/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java @@ -41,7 +41,7 @@ public interface ListInterface { @Schema( title = "Maximum number of objects returned across all paginated S3 calls", - description = "Total upper bound on the objects returned. It also sets the per-page size sent to S3, which AWS caps at 1000. The listing pages until this total is reached or the bucket is exhausted. Default 1000." + description = "Total upper bound on the objects returned, counted after `regexp` and `filter`. When `regexp` is unset and `filter` is `BOTH`, it also sets the per-page size sent to S3, which AWS caps at 1000. The listing pages until this total is reached or the bucket is exhausted. Default 1000." ) @PluginProperty(group = "execution") Property getMaxKeys(); diff --git a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java index 1d4c7881..804456a1 100644 --- a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java +++ b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java @@ -7,6 +7,7 @@ import java.util.HashMap; import java.util.List; import java.util.concurrent.ExecutionException; +import java.util.regex.Pattern; import org.apache.commons.io.FilenameUtils; import org.apache.commons.lang3.StringUtils; @@ -32,7 +33,6 @@ import software.amazon.awssdk.transfer.s3.model.DownloadFileRequest; import software.amazon.awssdk.transfer.s3.model.FileDownload; - public class S3Service { public static void initCrt() { @@ -239,11 +239,12 @@ static void performAction( public static List list(RunContext runContext, S3Client client, ListInterface list, AbstractS3Object abstractS3) throws IllegalVariableEvaluationException { int rMaxKeys = runContext.render(list.getMaxKeys()).as(Integer.class).orElse(1000); + if (rMaxKeys <= 0) { + return List.of(); + } ListObjectsV2Request.Builder builder = ListObjectsV2Request.builder() - .bucket(runContext.render(list.getBucket()).as(String.class).orElseThrow()) - // per-page size is capped at 1000 by S3, rMaxKeys is the total cap across pages - .maxKeys(Math.min(rMaxKeys, 1000)); + .bucket(runContext.render(list.getBucket()).as(String.class).orElseThrow(() -> new IllegalArgumentException("The 'bucket' property is required"))); if (list.getPrefix() != null) { builder.prefix(runContext.render(list.getPrefix()).as(String.class).orElseThrow()); @@ -272,16 +273,20 @@ public static List list(RunContext runContext, S3Client client, ListIn String regExp = runContext.render(list.getRegexp()).as(String.class).orElse(null); var filter = runContext.render(list.getFilter()).as(ListInterface.Filter.class).orElseThrow(); + var pattern = regExp == null ? null : Pattern.compile(regExp); + + // per-page size is capped at 1000 by S3, rMaxKeys is the total cap across pages, and a regexp or filter can skip most keys so it reads full pages + builder.maxKeys(pattern == null && filter == ListInterface.Filter.BOTH ? Math.min(rMaxKeys, 1000) : 1000); ListObjectsV2Request baseRequest = builder.build(); - List result = new ArrayList<>(); + var result = new ArrayList(); ListObjectsV2Request currentRequest = baseRequest; ListObjectsV2Response response; do { response = client.listObjectsV2(currentRequest); for (var s3Object : response.contents()) { - if (S3Service.filter(s3Object, regExp, filter)) { + if (S3Service.filter(s3Object, pattern, filter)) { result.add(S3Object.of(s3Object)); if (result.size() >= rMaxKeys) { return result; @@ -292,6 +297,7 @@ public static List list(RunContext runContext, S3Client client, ListIn if (response.isTruncated()) { String nextToken = response.nextContinuationToken(); if (nextToken == null) { + // Guard: isTruncated=true but no token returned, stop to avoid an infinite loop. runContext.logger().warn("S3 list pagination stopped: isTruncated=true but no continuation token available."); break; } @@ -302,8 +308,8 @@ public static List list(RunContext runContext, S3Client client, ListIn return result; } - private static boolean filter(software.amazon.awssdk.services.s3.model.S3Object object, String regExp, ListInterface.Filter filter) { - return (regExp == null || object.key().matches(regExp)) && + private static boolean filter(software.amazon.awssdk.services.s3.model.S3Object object, Pattern pattern, ListInterface.Filter filter) { + return (pattern == null || pattern.matcher(object.key()).matches()) && (filter == ListInterface.Filter.BOTH || (filter == ListInterface.Filter.DIRECTORY && object.key().endsWith("/")) || (filter == ListInterface.Filter.FILES && !object.key().endsWith("/"))); diff --git a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java index ceaf5482..e32b242d 100644 --- a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java +++ b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java @@ -8,6 +8,7 @@ import io.kestra.core.models.property.Property; import io.kestra.core.runners.RunContext; import io.kestra.core.runners.RunContextFactory; +import io.kestra.core.utils.IdUtils; import io.kestra.plugin.aws.s3.models.S3Object; import jakarta.inject.Inject; @@ -17,6 +18,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.*; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.*; @KestraTest @@ -77,7 +79,7 @@ void list_paginatesAcrossTwoTruncatedPages() throws Exception { @Test void shouldContinuePaginationWhenInitialPageDoesNotMatchFilter() throws Exception { - RunContext runContext = runContextFactory.of(); + var runContext = runContextFactory.of(); var obj1 = software.amazon.awssdk.services.s3.model.S3Object.builder() .key("archive/old1.txt").size(10L).build(); @@ -97,7 +99,7 @@ void shouldContinuePaginationWhenInitialPageDoesNotMatchFilter() throws Exceptio .contents(obj3) .build(); - S3Client mockClient = mock(S3Client.class); + var mockClient = mock(S3Client.class); when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) .thenAnswer(inv -> { ListObjectsV2Request req = inv.getArgument(0); @@ -105,23 +107,24 @@ void shouldContinuePaginationWhenInitialPageDoesNotMatchFilter() throws Exceptio }); var task = io.kestra.plugin.aws.s3.List.builder() - .id("s3-service-list-pagination-filter") + .id("s3-service-list-pagination-filter-%s".formatted(IdUtils.create())) .type(io.kestra.plugin.aws.s3.List.class.getName()) .bucket(Property.ofValue("my-bucket")) .regexp(Property.ofValue(".*\\.csv")) .maxKeys(Property.ofValue(2)) .build(); - List result = S3Service.list(runContext, mockClient, task, task); + var result = S3Service.list(runContext, mockClient, task, task); assertThat(result, hasSize(1)); - assertThat(result.get(0).getKey(), is("data/orders.csv")); - verify(mockClient, times(2)).listObjectsV2(any(ListObjectsV2Request.class)); + assertThat(result.getFirst().getKey(), is("data/orders.csv")); + // a regExp can skip most keys, so pages are requested at the S3 maximum instead of maxKeys + verify(mockClient, times(2)).listObjectsV2(argThat((ListObjectsV2Request req) -> req.maxKeys() == 1000)); } @Test void shouldStopPaginationWhenMatchingCountReachesMaxKeys() throws Exception { - RunContext runContext = runContextFactory.of(); + var runContext = runContextFactory.of(); var obj1 = software.amazon.awssdk.services.s3.model.S3Object.builder() .key("orders-1.csv").size(10L).build(); @@ -141,7 +144,7 @@ void shouldStopPaginationWhenMatchingCountReachesMaxKeys() throws Exception { .contents(obj3) .build(); - S3Client mockClient = mock(S3Client.class); + var mockClient = mock(S3Client.class); when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) .thenAnswer(inv -> { ListObjectsV2Request req = inv.getArgument(0); @@ -149,18 +152,51 @@ void shouldStopPaginationWhenMatchingCountReachesMaxKeys() throws Exception { }); var task = io.kestra.plugin.aws.s3.List.builder() - .id("s3-service-list-max-keys") + .id("s3-service-list-max-keys-%s".formatted(IdUtils.create())) .type(io.kestra.plugin.aws.s3.List.class.getName()) .bucket(Property.ofValue("my-bucket")) .regexp(Property.ofValue(".*\\.csv")) .maxKeys(Property.ofValue(2)) .build(); - List result = S3Service.list(runContext, mockClient, task, task); + var result = S3Service.list(runContext, mockClient, task, task); assertThat(result, hasSize(2)); - assertThat(result.get(0).getKey(), is("orders-1.csv")); + assertThat(result.getFirst().getKey(), is("orders-1.csv")); assertThat(result.get(1).getKey(), is("orders-2.csv")); verify(mockClient, times(1)).listObjectsV2(any(ListObjectsV2Request.class)); } + + @Test + void shouldRequestMaxKeysSizedPagesWithoutFilter() throws Exception { + var mockClient = mock(S3Client.class); + when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().isTruncated(false).build()); + var task = io.kestra.plugin.aws.s3.List.builder() + .id("s3-service-list-page-size-%s".formatted(IdUtils.create())) + .type(io.kestra.plugin.aws.s3.List.class.getName()) + .bucket(Property.ofValue("my-bucket")) + .maxKeys(Property.ofValue(5)) + .build(); + + S3Service.list(runContextFactory.of(), mockClient, task, task); + + verify(mockClient).listObjectsV2(argThat((ListObjectsV2Request req) -> req.maxKeys() == 5)); + } + + @Test + void shouldReturnNothingWithoutCallingS3WhenMaxKeysIsZero() throws Exception { + var mockClient = mock(S3Client.class); + var task = io.kestra.plugin.aws.s3.List.builder() + .id("s3-service-list-zero-max-keys-%s".formatted(IdUtils.create())) + .type(io.kestra.plugin.aws.s3.List.class.getName()) + .bucket(Property.ofValue("my-bucket")) + .maxKeys(Property.ofValue(0)) + .build(); + + var result = S3Service.list(runContextFactory.of(), mockClient, task, task); + + assertThat(result, empty()); + verify(mockClient, never()).listObjectsV2(any(ListObjectsV2Request.class)); + } } From ac19d8b3d48f5c20fcf1c6ccb5eefba1ec580000 Mon Sep 17 00:00:00 2001 From: Malay Dewangan Date: Mon, 26 Oct 2026 10:47:08 +0530 Subject: [PATCH 3/4] refactor(s3): filter each page with a stream, reject negative maxKeys and share test setup --- .../kestra/plugin/aws/s3/ListInterface.java | 5 +- .../io/kestra/plugin/aws/s3/S3Service.java | 28 +-- .../plugin/aws/s3/S3ServiceListTest.java | 175 ++++++++---------- 3 files changed, 92 insertions(+), 116 deletions(-) diff --git a/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java b/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java index 28c8a303..5b952c6d 100644 --- a/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java +++ b/src/main/java/io/kestra/plugin/aws/s3/ListInterface.java @@ -41,7 +41,10 @@ public interface ListInterface { @Schema( title = "Maximum number of objects returned across all paginated S3 calls", - description = "Total upper bound on the objects returned, counted after `regexp` and `filter`. When `regexp` is unset and `filter` is `BOTH`, it also sets the per-page size sent to S3, which AWS caps at 1000. The listing pages until this total is reached or the bucket is exhausted. Default 1000." + description = """ + Total upper bound on the objects returned, counted after `regexp` and `filter`. Default 1000. + When `regexp` is unset and `filter` is `BOTH`, it also sets the per-page size sent to S3, which AWS caps at 1000. + The listing pages until this total is reached or the bucket is exhausted.""" ) @PluginProperty(group = "execution") Property getMaxKeys(); diff --git a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java index 804456a1..82c26e43 100644 --- a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java +++ b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java @@ -239,12 +239,15 @@ static void performAction( public static List list(RunContext runContext, S3Client client, ListInterface list, AbstractS3Object abstractS3) throws IllegalVariableEvaluationException { int rMaxKeys = runContext.render(list.getMaxKeys()).as(Integer.class).orElse(1000); - if (rMaxKeys <= 0) { + if (rMaxKeys < 0) { + throw new IllegalArgumentException("The `maxKeys` property must be 0 or more, got %d.".formatted(rMaxKeys)); + } + if (rMaxKeys == 0) { return List.of(); } ListObjectsV2Request.Builder builder = ListObjectsV2Request.builder() - .bucket(runContext.render(list.getBucket()).as(String.class).orElseThrow(() -> new IllegalArgumentException("The 'bucket' property is required"))); + .bucket(runContext.render(list.getBucket()).as(String.class).orElseThrow(() -> new IllegalArgumentException("The `bucket` property is required."))); if (list.getPrefix() != null) { builder.prefix(runContext.render(list.getPrefix()).as(String.class).orElseThrow()); @@ -271,12 +274,11 @@ public static List list(RunContext runContext, S3Client client, ListIn builder.requestPayer(runContext.render(abstractS3.getRequestPayer()).as(String.class).orElseThrow()); } - String regExp = runContext.render(list.getRegexp()).as(String.class).orElse(null); - var filter = runContext.render(list.getFilter()).as(ListInterface.Filter.class).orElseThrow(); - var pattern = regExp == null ? null : Pattern.compile(regExp); + var rPattern = runContext.render(list.getRegexp()).as(String.class).map(Pattern::compile).orElse(null); + var rFilter = runContext.render(list.getFilter()).as(ListInterface.Filter.class).orElse(ListInterface.Filter.BOTH); // per-page size is capped at 1000 by S3, rMaxKeys is the total cap across pages, and a regexp or filter can skip most keys so it reads full pages - builder.maxKeys(pattern == null && filter == ListInterface.Filter.BOTH ? Math.min(rMaxKeys, 1000) : 1000); + builder.maxKeys(rPattern == null && rFilter == ListInterface.Filter.BOTH ? Math.min(rMaxKeys, 1000) : 1000); ListObjectsV2Request baseRequest = builder.build(); var result = new ArrayList(); @@ -285,13 +287,13 @@ public static List list(RunContext runContext, S3Client client, ListIn ListObjectsV2Response response; do { response = client.listObjectsV2(currentRequest); - for (var s3Object : response.contents()) { - if (S3Service.filter(s3Object, pattern, filter)) { - result.add(S3Object.of(s3Object)); - if (result.size() >= rMaxKeys) { - return result; - } - } + response.contents().stream() + .filter(s3Object -> filter(s3Object, rPattern, rFilter)) + .limit(rMaxKeys - result.size()) + .map(S3Object::of) + .forEach(result::add); + if (result.size() >= rMaxKeys) { + return result; } if (response.isTruncated()) { diff --git a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java index e32b242d..bb2be412 100644 --- a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java +++ b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java @@ -17,6 +17,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.*; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.*; @@ -32,36 +33,9 @@ class S3ServiceListTest { void list_paginatesAcrossTwoTruncatedPages() throws Exception { RunContext runContext = runContextFactory.of(); - var obj1 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("file1.txt").size(10L).build(); - var obj2 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("file2.txt").size(20L).build(); - var obj3 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("file3.txt").size(30L).build(); + var mockClient = twoPageClient(List.of("file1.txt", "file2.txt"), List.of("file3.txt")); - var page1 = ListObjectsV2Response.builder() - .isTruncated(true) - .nextContinuationToken("token-page2") - .contents(obj1, obj2) - .build(); - - var page2 = ListObjectsV2Response.builder() - .isTruncated(false) - .contents(obj3) - .build(); - - S3Client mockClient = mock(S3Client.class); - when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) - .thenAnswer(inv -> - { - ListObjectsV2Request req = inv.getArgument(0); - return "token-page2".equals(req.continuationToken()) ? page2 : page1; - }); - - var task = io.kestra.plugin.aws.s3.List.builder() - .id("s3-service-list-unit-test") - .type(io.kestra.plugin.aws.s3.List.class.getName()) - .bucket(Property.ofValue("my-bucket")) + var task = listTask() .maxKeys(Property.ofValue(1000)) .maxFiles(Property.ofValue(1000)) .build(); @@ -79,92 +53,50 @@ void list_paginatesAcrossTwoTruncatedPages() throws Exception { @Test void shouldContinuePaginationWhenInitialPageDoesNotMatchFilter() throws Exception { - var runContext = runContextFactory.of(); - - var obj1 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("archive/old1.txt").size(10L).build(); - var obj2 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("archive/old2.txt").size(20L).build(); - var obj3 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("data/orders.csv").size(30L).build(); - - var page1 = ListObjectsV2Response.builder() - .isTruncated(true) - .nextContinuationToken("token-page2") - .contents(obj1, obj2) - .build(); + var mockClient = twoPageClient(List.of("archive/old1.txt", "archive/old2.txt"), List.of("data/orders.csv")); - var page2 = ListObjectsV2Response.builder() - .isTruncated(false) - .contents(obj3) - .build(); - - var mockClient = mock(S3Client.class); - when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) - .thenAnswer(inv -> { - ListObjectsV2Request req = inv.getArgument(0); - return "token-page2".equals(req.continuationToken()) ? page2 : page1; - }); - - var task = io.kestra.plugin.aws.s3.List.builder() - .id("s3-service-list-pagination-filter-%s".formatted(IdUtils.create())) - .type(io.kestra.plugin.aws.s3.List.class.getName()) - .bucket(Property.ofValue("my-bucket")) + var task = listTask() .regexp(Property.ofValue(".*\\.csv")) .maxKeys(Property.ofValue(2)) .build(); - var result = S3Service.list(runContext, mockClient, task, task); + var result = S3Service.list(runContextFactory.of(), mockClient, task, task); assertThat(result, hasSize(1)); assertThat(result.getFirst().getKey(), is("data/orders.csv")); - // a regExp can skip most keys, so pages are requested at the S3 maximum instead of maxKeys + // a regexp can skip most keys, so pages are requested at the S3 maximum instead of maxKeys verify(mockClient, times(2)).listObjectsV2(argThat((ListObjectsV2Request req) -> req.maxKeys() == 1000)); } @Test void shouldStopPaginationWhenMatchingCountReachesMaxKeys() throws Exception { - var runContext = runContextFactory.of(); - - var obj1 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("orders-1.csv").size(10L).build(); - var obj2 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("orders-2.csv").size(20L).build(); - var obj3 = software.amazon.awssdk.services.s3.model.S3Object.builder() - .key("orders-3.csv").size(30L).build(); + var mockClient = twoPageClient(List.of("orders-1.csv", "notes.txt"), List.of("orders-2.csv", "orders-3.csv")); - var page1 = ListObjectsV2Response.builder() - .isTruncated(true) - .nextContinuationToken("token-page2") - .contents(obj1, obj2) + var task = listTask() + .regexp(Property.ofValue(".*\\.csv")) + .maxKeys(Property.ofValue(2)) .build(); - var page2 = ListObjectsV2Response.builder() - .isTruncated(false) - .contents(obj3) - .build(); + var result = S3Service.list(runContextFactory.of(), mockClient, task, task); - var mockClient = mock(S3Client.class); - when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) - .thenAnswer(inv -> { - ListObjectsV2Request req = inv.getArgument(0); - return "token-page2".equals(req.continuationToken()) ? page2 : page1; - }); + assertThat(result.stream().map(S3Object::getKey).toList(), contains("orders-1.csv", "orders-2.csv")); + verify(mockClient, times(2)).listObjectsV2(any(ListObjectsV2Request.class)); + } - var task = io.kestra.plugin.aws.s3.List.builder() - .id("s3-service-list-max-keys-%s".formatted(IdUtils.create())) - .type(io.kestra.plugin.aws.s3.List.class.getName()) - .bucket(Property.ofValue("my-bucket")) - .regexp(Property.ofValue(".*\\.csv")) - .maxKeys(Property.ofValue(2)) + @Test + void shouldContinuePaginationWhenInitialPageHasOnlyDirectories() throws Exception { + var mockClient = twoPageClient(List.of("archive/", "data/"), List.of("data/orders.csv")); + + var task = listTask() + .filter(Property.ofValue(ListInterface.Filter.FILES)) + .maxKeys(Property.ofValue(1)) .build(); - var result = S3Service.list(runContext, mockClient, task, task); + var result = S3Service.list(runContextFactory.of(), mockClient, task, task); - assertThat(result, hasSize(2)); - assertThat(result.getFirst().getKey(), is("orders-1.csv")); - assertThat(result.get(1).getKey(), is("orders-2.csv")); - verify(mockClient, times(1)).listObjectsV2(any(ListObjectsV2Request.class)); + assertThat(result, hasSize(1)); + assertThat(result.getFirst().getKey(), is("data/orders.csv")); + verify(mockClient, times(2)).listObjectsV2(argThat((ListObjectsV2Request req) -> req.maxKeys() == 1000)); } @Test @@ -172,10 +104,7 @@ void shouldRequestMaxKeysSizedPagesWithoutFilter() throws Exception { var mockClient = mock(S3Client.class); when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) .thenReturn(ListObjectsV2Response.builder().isTruncated(false).build()); - var task = io.kestra.plugin.aws.s3.List.builder() - .id("s3-service-list-page-size-%s".formatted(IdUtils.create())) - .type(io.kestra.plugin.aws.s3.List.class.getName()) - .bucket(Property.ofValue("my-bucket")) + var task = listTask() .maxKeys(Property.ofValue(5)) .build(); @@ -187,10 +116,7 @@ void shouldRequestMaxKeysSizedPagesWithoutFilter() throws Exception { @Test void shouldReturnNothingWithoutCallingS3WhenMaxKeysIsZero() throws Exception { var mockClient = mock(S3Client.class); - var task = io.kestra.plugin.aws.s3.List.builder() - .id("s3-service-list-zero-max-keys-%s".formatted(IdUtils.create())) - .type(io.kestra.plugin.aws.s3.List.class.getName()) - .bucket(Property.ofValue("my-bucket")) + var task = listTask() .maxKeys(Property.ofValue(0)) .build(); @@ -199,4 +125,49 @@ void shouldReturnNothingWithoutCallingS3WhenMaxKeysIsZero() throws Exception { assertThat(result, empty()); verify(mockClient, never()).listObjectsV2(any(ListObjectsV2Request.class)); } + + @Test + void shouldRejectNegativeMaxKeys() { + var mockClient = mock(S3Client.class); + var task = listTask() + .maxKeys(Property.ofValue(-1)) + .build(); + + var exception = assertThrows(IllegalArgumentException.class, () -> S3Service.list(runContextFactory.of(), mockClient, task, task)); + + assertThat(exception.getMessage(), containsString("must be 0 or more")); + verify(mockClient, never()).listObjectsV2(any(ListObjectsV2Request.class)); + } + + private static io.kestra.plugin.aws.s3.List.ListBuilder listTask() { + return io.kestra.plugin.aws.s3.List.builder() + .id("s3-service-list-%s".formatted(IdUtils.create())) + .type(io.kestra.plugin.aws.s3.List.class.getName()) + .bucket(Property.ofValue("my-bucket")); + } + + // Serves firstPage, then secondPage once the request carries the continuation token. + private static S3Client twoPageClient(List firstPage, List secondPage) { + var page1 = ListObjectsV2Response.builder() + .isTruncated(true) + .nextContinuationToken("token-page2") + .contents(s3Objects(firstPage)) + .build(); + + var page2 = ListObjectsV2Response.builder() + .isTruncated(false) + .contents(s3Objects(secondPage)) + .build(); + + var mockClient = mock(S3Client.class); + when(mockClient.listObjectsV2(any(ListObjectsV2Request.class))) + .thenAnswer(inv -> "token-page2".equals(inv.getArgument(0).continuationToken()) ? page2 : page1); + return mockClient; + } + + private static List s3Objects(List keys) { + return keys.stream() + .map(key -> software.amazon.awssdk.services.s3.model.S3Object.builder().key(key).size(10L).build()) + .toList(); + } } From ac07c6510b54c96a180846daf5af7eda244cf712 Mon Sep 17 00:00:00 2001 From: Malay Dewangan Date: Mon, 26 Oct 2026 10:54:44 +0530 Subject: [PATCH 4/4] docs(s3): document maxKeys bounds and scan cost, rename the filter type parameter --- src/main/java/io/kestra/plugin/aws/s3/Download.java | 2 +- .../java/io/kestra/plugin/aws/s3/ListInterface.java | 4 ++-- src/main/java/io/kestra/plugin/aws/s3/S3Service.java | 8 ++++---- .../io/kestra/plugin/aws/s3/S3ServiceListTest.java | 10 ++-------- 4 files changed, 9 insertions(+), 15 deletions(-) diff --git a/src/main/java/io/kestra/plugin/aws/s3/Download.java b/src/main/java/io/kestra/plugin/aws/s3/Download.java index dbb3a5e2..f06058e8 100644 --- a/src/main/java/io/kestra/plugin/aws/s3/Download.java +++ b/src/main/java/io/kestra/plugin/aws/s3/Download.java @@ -126,7 +126,7 @@ public class Download extends AbstractS3Object implements RunnableTask getMaxKeys(); diff --git a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java index 82c26e43..5b700bd2 100644 --- a/src/main/java/io/kestra/plugin/aws/s3/S3Service.java +++ b/src/main/java/io/kestra/plugin/aws/s3/S3Service.java @@ -310,10 +310,10 @@ public static List list(RunContext runContext, S3Client client, ListIn return result; } - private static boolean filter(software.amazon.awssdk.services.s3.model.S3Object object, Pattern pattern, ListInterface.Filter filter) { + private static boolean filter(software.amazon.awssdk.services.s3.model.S3Object object, Pattern pattern, ListInterface.Filter type) { return (pattern == null || pattern.matcher(object.key()).matches()) && - (filter == ListInterface.Filter.BOTH || - (filter == ListInterface.Filter.DIRECTORY && object.key().endsWith("/")) || - (filter == ListInterface.Filter.FILES && !object.key().endsWith("/"))); + (type == ListInterface.Filter.BOTH || + (type == ListInterface.Filter.DIRECTORY && object.key().endsWith("/")) || + (type == ListInterface.Filter.FILES && !object.key().endsWith("/"))); } } diff --git a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java index bb2be412..cc505dbe 100644 --- a/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java +++ b/src/test/java/io/kestra/plugin/aws/s3/S3ServiceListTest.java @@ -6,7 +6,6 @@ import io.kestra.core.junit.annotations.KestraTest; import io.kestra.core.models.property.Property; -import io.kestra.core.runners.RunContext; import io.kestra.core.runners.RunContextFactory; import io.kestra.core.utils.IdUtils; import io.kestra.plugin.aws.s3.models.S3Object; @@ -31,8 +30,6 @@ class S3ServiceListTest { // Verifies the V2 isTruncated -> continuationToken -> next page loop. @Test void list_paginatesAcrossTwoTruncatedPages() throws Exception { - RunContext runContext = runContextFactory.of(); - var mockClient = twoPageClient(List.of("file1.txt", "file2.txt"), List.of("file3.txt")); var task = listTask() @@ -40,12 +37,9 @@ void list_paginatesAcrossTwoTruncatedPages() throws Exception { .maxFiles(Property.ofValue(1000)) .build(); - List result = S3Service.list(runContext, mockClient, task, task); + var result = S3Service.list(runContextFactory.of(), mockClient, task, task); - assertThat(result, hasSize(3)); - assertThat(result.get(0).getKey(), is("file1.txt")); - assertThat(result.get(1).getKey(), is("file2.txt")); - assertThat(result.get(2).getKey(), is("file3.txt")); + assertThat(result.stream().map(S3Object::getKey).toList(), contains("file1.txt", "file2.txt", "file3.txt")); // Verify the second call carried the continuation token. verify(mockClient, times(2)).listObjectsV2(any(ListObjectsV2Request.class));