This is an automated email from the ASF dual-hosted git repository.
SbloodyS pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 8592557809 [Fix-18378] Fix list resources returned only 1000 records
in s3 storage type (#18381)
8592557809 is described below
commit 8592557809d2d041d9d727371c090aef82e4dcae
Author: xiangzihao <[email protected]>
AuthorDate: Fri Jun 26 11:35:19 2026 +0800
[Fix-18378] Fix list resources returned only 1000 records in s3 storage
type (#18381)
---
.../plugin/storage/s3/S3StorageOperator.java | 35 +++++++++------
.../plugin/storage/s3/S3StorageOperatorTest.java | 52 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 13 deletions(-)
diff --git
a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java
b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java
index 2f4954fe1c..df9b123f2b 100644
---
a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java
+++
b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java
@@ -65,12 +65,17 @@ public class S3StorageOperator extends
AbstractStorageOperator implements Closea
private final AmazonS3 s3Client;
public S3StorageOperator(S3StorageProperties s3StorageProperties) {
- super(s3StorageProperties.getResourceUploadPath());
- bucketName = s3StorageProperties.getBucketName();
- s3Client =
AmazonS3ClientFactory.createAmazonS3Client(s3StorageProperties.getS3Configuration());
+ this(s3StorageProperties.getBucketName(),
s3StorageProperties.getResourceUploadPath(),
+
AmazonS3ClientFactory.createAmazonS3Client(s3StorageProperties.getS3Configuration()));
exceptionWhenBucketNameNotExists(bucketName);
}
+ S3StorageOperator(String bucketName, String resourceUploadPath, AmazonS3
s3Client) {
+ super(resourceUploadPath);
+ this.bucketName = bucketName;
+ this.s3Client = s3Client;
+ }
+
@Override
public String getStorageBaseDirectory() {
// All directory should end with File.separator
@@ -224,17 +229,21 @@ public class S3StorageOperator extends
AbstractStorageOperator implements Closea
.withDelimiter("/")
.withPrefix(s3ResourceAbsolutePath);
- ListObjectsV2Result listObjectsV2Result =
s3Client.listObjectsV2(listObjectsV2Request);
List<StorageEntity> storageEntities = new ArrayList<>();
- storageEntities.addAll(listObjectsV2Result.getCommonPrefixes()
- .stream()
- .map(this::transformCommonPrefixToStorageEntity)
- .collect(Collectors.toList()));
- storageEntities.addAll(
- listObjectsV2Result.getObjectSummaries().stream()
- .filter(s3ObjectSummary ->
!s3ObjectSummary.getKey().equals(s3ResourceAbsolutePath))
- .map(this::transformS3ObjectToStorageEntity)
- .collect(Collectors.toList()));
+ ListObjectsV2Result listObjectsV2Result;
+ do {
+ listObjectsV2Result = s3Client.listObjectsV2(listObjectsV2Request);
+ storageEntities.addAll(listObjectsV2Result.getCommonPrefixes()
+ .stream()
+ .map(this::transformCommonPrefixToStorageEntity)
+ .collect(Collectors.toList()));
+ storageEntities.addAll(
+ listObjectsV2Result.getObjectSummaries().stream()
+ .filter(s3ObjectSummary ->
!s3ObjectSummary.getKey().equals(s3ResourceAbsolutePath))
+ .map(this::transformS3ObjectToStorageEntity)
+ .collect(Collectors.toList()));
+
listObjectsV2Request.setContinuationToken(listObjectsV2Result.getNextContinuationToken());
+ } while (listObjectsV2Result.isTruncated());
return storageEntities;
}
diff --git
a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/test/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperatorTest.java
b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/test/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperatorTest.java
index dc635cceb5..e5e5ca0aab 100644
---
a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/test/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperatorTest.java
+++
b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/test/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperatorTest.java
@@ -27,8 +27,10 @@ import
org.apache.dolphinscheduler.plugin.storage.api.ResourceMetadata;
import org.apache.dolphinscheduler.plugin.storage.api.StorageEntity;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
+import java.lang.reflect.Proxy;
import java.nio.file.FileAlreadyExistsException;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import lombok.SneakyThrows;
@@ -46,6 +48,9 @@ import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.ListObjectsV2Request;
+import com.amazonaws.services.s3.model.ListObjectsV2Result;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
public class S3StorageOperatorTest {
@@ -259,6 +264,20 @@ public class S3StorageOperatorTest {
}
+ @Test
+ public void testListStorageEntity_shouldFetchAllS3Pages() {
+ AtomicInteger listObjectsCallCount = new AtomicInteger();
+ AmazonS3 amazonS3 = createPagedAmazonS3Client(listObjectsCallCount);
+ S3StorageOperator storageOperator = new
S3StorageOperator("dolphinscheduler",
+ "tmp/dolphinscheduler", amazonS3);
+
+ List<StorageEntity> storageEntities =
+
storageOperator.listStorageEntity("tmp/dolphinscheduler/default/resources/largeDirectory");
+
+ assertThat(storageEntities).hasSize(1001);
+ assertThat(listObjectsCallCount.get()).isEqualTo(2);
+ }
+
@Test
public void testListStorageEntity_directoryNotExist() {
List<StorageEntity> storageEntities =
@@ -302,4 +321,37 @@ public class S3StorageOperatorTest {
}
}
+ private AmazonS3 createPagedAmazonS3Client(AtomicInteger
listObjectsCallCount) {
+ return (AmazonS3) Proxy.newProxyInstance(
+ AmazonS3.class.getClassLoader(),
+ new Class<?>[]{AmazonS3.class},
+ (proxy, method, args) -> {
+ if (!"listObjectsV2".equals(method.getName())) {
+ return null;
+ }
+
+ ListObjectsV2Request request = (ListObjectsV2Request)
args[0];
+ int callIndex = listObjectsCallCount.getAndIncrement();
+ if (callIndex == 0) {
+ assertThat(request.getContinuationToken()).isNull();
+ return createListObjectsV2Result("next-page-token", 0,
1000);
+ }
+
assertThat(request.getContinuationToken()).isEqualTo("next-page-token");
+ return createListObjectsV2Result(null, 1000, 1001);
+ });
+ }
+
+ private ListObjectsV2Result createListObjectsV2Result(String
nextContinuationToken, int start, int end) {
+ ListObjectsV2Result listObjectsV2Result = new ListObjectsV2Result();
+ listObjectsV2Result.setTruncated(nextContinuationToken != null);
+ listObjectsV2Result.setNextContinuationToken(nextContinuationToken);
+ for (int i = start; i < end; i++) {
+ S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
+ s3ObjectSummary.setKey(String.format(
+
"tmp/dolphinscheduler/default/resources/largeDirectory/demo-%04d.sql", i));
+ listObjectsV2Result.getObjectSummaries().add(s3ObjectSummary);
+ }
+ return listObjectsV2Result;
+ }
+
}
