This is an automated email from the ASF dual-hosted git repository.
Samrat002 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ca17e52868e [FLINK-40593][s3] Batch recursive S3 deletes in native s3
connector with global and per-bucket opt-out
ca17e52868e is described below
commit ca17e52868e2da8995c9d7a2dcb942c083a28bc1
Author: Gabor Somogyi <[email protected]>
AuthorDate: Thu Sep 10 13:05:33 2026 +0200
[FLINK-40593][s3] Batch recursive S3 deletes in native s3 connector with
global and per-bucket opt-out
* [FLINK-40593][s3] Batch recursive S3 deletes in native s3 connector with
global and per-bucket opt-out
---
flink-filesystems/flink-s3-fs-native/README.md | 2 +
.../flink/fs/s3native/BucketConfigProvider.java | 12 ++
.../flink/fs/s3native/NativeS3FileSystem.java | 22 ++-
.../fs/s3native/NativeS3FileSystemFactory.java | 17 +-
.../flink/fs/s3native/NativeS3RecursiveDelete.java | 150 +++++++++++++++++
.../apache/flink/fs/s3native/S3BucketConfig.java | 25 ++-
.../fs/s3native/BucketConfigProviderTest.java | 15 +-
.../fs/s3native/NativeS3FileSystemFactoryTest.java | 23 +++
.../fs/s3native/NativeS3FileSystemITCase.java | 33 +++-
.../fs/s3native/NativeS3RecursiveDeleteTest.java | 182 +++++++++++++++++++++
.../src/test/resources/log4j2-test.properties | 28 ++++
11 files changed, 495 insertions(+), 14 deletions(-)
diff --git a/flink-filesystems/flink-s3-fs-native/README.md
b/flink-filesystems/flink-s3-fs-native/README.md
index b65583f989d..7f45ffe13ab 100644
--- a/flink-filesystems/flink-s3-fs-native/README.md
+++ b/flink-filesystems/flink-s3-fs-native/README.md
@@ -77,6 +77,7 @@ input.sinkTo(FileSink.forRowFormat(new
Path("s3://my-bucket/output"),
| s3.connection.max | 50 | Maximum HTTP connections in the S3 client
connection pool. Applies to sync and async clients, including CRT when enabled.
Must be ≥ `s3.bulk-copy.max-concurrent` |
| s3.async.enabled | true | Enable async read/write with TransferManager |
| s3.read.buffer.size | 262144 (256KB) | Read buffer size per stream (64KB -
4MB) |
+| s3.delete.batch.enabled | true | Use S3's batch `DeleteObjects` API when
recursively deleting a directory, instead of issuing one `DeleteObject` call
per file. Disable for S3-compatible stores that do not support multi-object
delete |
### Metrics
@@ -156,6 +157,7 @@ Only the following properties can be overridden at the
bucket level. Any other `
- **Credentials:** `access-key`, `secret-key`, `aws.credentials.provider`
- **Encryption:** `sse.type`, `sse.kms.key-id`
- **IAM Assume Role:** `assume-role.arn`, `assume-role.external-id`,
`assume-role.session-name`, `assume-role.session-duration`
+- **Delete behavior:** `delete.batch.enabled`
Timeouts, retries, encoding/checksum flags, entropy, upload/copy settings, and
the credentials provider chain are configured globally only.
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/BucketConfigProvider.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/BucketConfigProvider.java
index 05355f8d4a0..248a4100b87 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/BucketConfigProvider.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/BucketConfigProvider.java
@@ -76,6 +76,18 @@ final class BucketConfigProvider {
});
applicators.put("assume-role.session-name",
S3BucketConfig.Builder::assumeRoleSessionName);
applicators.put("aws.credentials.provider",
S3BucketConfig.Builder::credentialsProvider);
+ applicators.put(
+ "delete.batch.enabled",
+ (b, v) -> {
+ if (!"true".equalsIgnoreCase(v) &&
!"false".equalsIgnoreCase(v)) {
+ throw new IllegalConfigurationException(
+ String.format(
+ "Invalid delete.batch.enabled '%s' for
bucket '%s'. "
+ + "Must be 'true' or 'false'",
+ v, b.getBucketName()));
+ }
+ b.deleteBatchEnabled(Boolean.parseBoolean(v));
+ });
applicators.put("endpoint", S3BucketConfig.Builder::endpoint);
applicators.put(
"path-style-access",
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystem.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystem.java
index efec0dcc185..ebb4a603a5d 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystem.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystem.java
@@ -114,6 +114,7 @@ class NativeS3FileSystem extends FileSystem
private final boolean useAsyncOperations;
private final int readBufferSize;
private final Duration fsCloseTimeout;
+ private final boolean deleteBatchEnabled;
private final AtomicBoolean closed = new AtomicBoolean(false);
public NativeS3FileSystem(
@@ -127,7 +128,8 @@ class NativeS3FileSystem extends FileSystem
@Nullable NativeS3BulkCopyHelper bulkCopyHelper,
boolean useAsyncOperations,
int readBufferSize,
- Duration fsCloseTimeout) {
+ Duration fsCloseTimeout,
+ boolean deleteBatchEnabled) {
this.clientProvider =
Preconditions.checkNotNull(clientProvider, "clientProvider
must not be null");
this.uri = uri;
@@ -140,6 +142,7 @@ class NativeS3FileSystem extends FileSystem
this.useAsyncOperations = useAsyncOperations;
this.readBufferSize = readBufferSize;
this.fsCloseTimeout = fsCloseTimeout;
+ this.deleteBatchEnabled = deleteBatchEnabled;
this.s3AccessHelper =
new NativeS3ObjectOperations(
clientProvider.getS3Client(),
@@ -155,11 +158,12 @@ class NativeS3FileSystem extends FileSystem
}
LOG.info(
- "Created Native S3 FileSystem for bucket: {}, entropy
injection: {}, bulk copy: {}, read buffer: {} KB",
+ "Created Native S3 FileSystem for bucket: {}, entropy
injection: {}, bulk copy: {}, read buffer: {} KB, delete batching: {}",
bucketName,
entropyInjectionKey != null,
bulkCopyHelper != null,
- readBufferSize / 1024);
+ readBufferSize / 1024,
+ deleteBatchEnabled);
}
@VisibleForTesting
@@ -167,6 +171,11 @@ class NativeS3FileSystem extends FileSystem
return fsCloseTimeout;
}
+ @VisibleForTesting
+ boolean isDeleteBatchEnabled() {
+ return deleteBatchEnabled;
+ }
+
@VisibleForTesting
S3ClientProvider getClientProvider() {
return clientProvider;
@@ -392,11 +401,8 @@ class NativeS3FileSystem extends FileSystem
throw new IOException("Directory not empty and recursive =
false");
}
- final FileStatus[] contents = listStatus(path);
- for (FileStatus file : contents) {
- delete(file.getPath(), true);
- }
-
+ new NativeS3RecursiveDelete(s3Client, bucketName, key,
deleteBatchEnabled)
+ .execute();
return true;
}
} catch (FileNotFoundException e) {
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
index ff52b2dfb0a..63f6ac118ee 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
@@ -188,6 +188,16 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory, MetricsAwar
.withDescription(
"Enable async read/write operations using
S3TransferManager for improved performance");
+ public static final ConfigOption<Boolean> DELETE_BATCH_ENABLED =
+ ConfigOptions.key("s3.delete.batch.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Use S3's batch DeleteObjects API when recursively
deleting a "
+ + "directory, instead of issuing one
DeleteObject call per "
+ + "file. Disable for S3-compatible stores
that don't support "
+ + "multi-object delete.");
+
public static final ConfigOption<Integer> READ_BUFFER_SIZE =
ConfigOptions.key("s3.read.buffer.size")
.intType()
@@ -529,6 +539,7 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory, MetricsAwar
String assumeRoleSessionName = config.get(ASSUME_ROLE_SESSION_NAME);
int assumeRoleSessionDuration =
config.get(ASSUME_ROLE_SESSION_DURATION_SECONDS);
String credentialsProviderClasses =
config.get(AWS_CREDENTIALS_PROVIDER);
+ boolean deleteBatchEnabled = config.get(DELETE_BATCH_ENABLED);
// Apply bucket-specific overrides
String bucketName = fsUri.getHost();
@@ -562,6 +573,9 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory, MetricsAwar
if (overrides.getAssumeRoleSessionDurationSeconds() != null) {
assumeRoleSessionDuration =
overrides.getAssumeRoleSessionDurationSeconds();
}
+ if (overrides.getDeleteBatchEnabled() != null) {
+ deleteBatchEnabled = overrides.getDeleteBatchEnabled();
+ }
}
}
@@ -750,7 +764,8 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory, MetricsAwar
bulkCopyHelper,
useAsyncOperations,
readBufferSize,
- config.get(FS_CLOSE_TIMEOUT));
+ config.get(FS_CLOSE_TIMEOUT),
+ deleteBatchEnabled);
}
@Nullable
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3RecursiveDelete.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3RecursiveDelete.java
new file mode 100644
index 00000000000..ba2c90c05b6
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3RecursiveDelete.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3native;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.Delete;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
+import software.amazon.awssdk.services.s3.model.S3Error;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A single recursive deletion of every object under one key prefix.
Instantiated per {@code
+ * delete(path, true)} call; walks the prefix one listing page at a time so
memory use stays bounded
+ * by the page size rather than the size of the whole subtree, since each page
maps directly onto
+ * one {@code DeleteObjects} request.
+ */
+final class NativeS3RecursiveDelete {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(NativeS3RecursiveDelete.class);
+
+ /** Also used as the listing page size (the S3 {@code DeleteObjects}
limit). */
+ private static final int PAGE_SIZE = 1000;
+
+ private final S3Client s3Client;
+ private final String bucketName;
+ private final String prefix;
+ private final boolean batchEnabled;
+
+ NativeS3RecursiveDelete(
+ S3Client s3Client, String bucketName, String key, boolean
batchEnabled) {
+ this.s3Client = s3Client;
+ this.bucketName = bucketName;
+ this.prefix = key.endsWith("/") ? key : key + "/";
+ this.batchEnabled = batchEnabled;
+ }
+
+ /** Deletes every object under the prefix using the configured delete
strategy. */
+ void execute() throws IOException {
+ String continuationToken = null;
+
+ do {
+ final ListObjectsV2Response response = listPage(continuationToken);
+ final List<String> pageKeys = extractKeys(response);
+
+ if (!pageKeys.isEmpty()) {
+ deletePage(pageKeys);
+ }
+
+ continuationToken = response.nextContinuationToken();
+ } while (continuationToken != null);
+ }
+
+ private ListObjectsV2Response listPage(String continuationToken) {
+ final ListObjectsV2Request.Builder requestBuilder =
+
ListObjectsV2Request.builder().bucket(bucketName).prefix(prefix).maxKeys(PAGE_SIZE);
+ if (continuationToken != null) {
+ requestBuilder.continuationToken(continuationToken);
+ }
+ return s3Client.listObjectsV2(requestBuilder.build());
+ }
+
+ private List<String> extractKeys(ListObjectsV2Response response) {
+ final List<String> keys = new ArrayList<>(response.contents().size());
+ for (S3Object s3Object : response.contents()) {
+ keys.add(s3Object.key());
+ }
+ return keys;
+ }
+
+ private void deletePage(List<String> keys) throws IOException {
+ if (batchEnabled) {
+ LOG.debug(
+ "Deleting {} object(s) under prefix {} using batched
DeleteObjects",
+ keys.size(),
+ prefix);
+ deleteBatch(keys);
+ } else {
+ LOG.debug(
+ "Deleting {} object(s) under prefix {} using individual
DeleteObject calls "
+ + "(delete batching disabled)",
+ keys.size(),
+ prefix);
+ deleteIndividually(keys);
+ }
+ }
+
+ private void deleteIndividually(List<String> keys) {
+ for (String key : keys) {
+ final DeleteObjectRequest request =
+
DeleteObjectRequest.builder().bucket(bucketName).key(key).build();
+ s3Client.deleteObject(request);
+ }
+ }
+
+ private void deleteBatch(List<String> keys) throws IOException {
+ final List<ObjectIdentifier> objectIdentifiers = new
ArrayList<>(keys.size());
+ for (String key : keys) {
+ objectIdentifiers.add(ObjectIdentifier.builder().key(key).build());
+ }
+
+ final DeleteObjectsRequest request =
+ DeleteObjectsRequest.builder()
+ .bucket(bucketName)
+
.delete(Delete.builder().objects(objectIdentifiers).build())
+ .build();
+
+ LOG.debug("Issuing batch DeleteObjects request for {} key(s)",
keys.size());
+ final DeleteObjectsResponse response = s3Client.deleteObjects(request);
+ if (response.hasErrors() && !response.errors().isEmpty()) {
+ final StringBuilder errorMessage = new StringBuilder("Failed to
delete objects: ");
+ for (S3Error error : response.errors()) {
+ errorMessage
+ .append(error.key())
+ .append(" (")
+ .append(error.code())
+ .append(": ")
+ .append(error.message())
+ .append("); ");
+ }
+ throw new IOException(errorMessage.toString());
+ }
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3BucketConfig.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3BucketConfig.java
index 0625254cd33..13f52e8cec9 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3BucketConfig.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3BucketConfig.java
@@ -51,6 +51,7 @@ final class S3BucketConfig {
@Nullable private final String assumeRoleSessionName;
@Nullable private final Integer assumeRoleSessionDurationSeconds;
@Nullable private final String credentialsProvider;
+ @Nullable private final Boolean deleteBatchEnabled;
private S3BucketConfig(Builder builder) {
this.bucketName = builder.bucketName;
@@ -66,6 +67,7 @@ final class S3BucketConfig {
this.assumeRoleSessionName = builder.assumeRoleSessionName;
this.assumeRoleSessionDurationSeconds =
builder.assumeRoleSessionDurationSeconds;
this.credentialsProvider = builder.credentialsProvider;
+ this.deleteBatchEnabled = builder.deleteBatchEnabled;
}
String getBucketName() {
@@ -132,6 +134,11 @@ final class S3BucketConfig {
return credentialsProvider;
}
+ @Nullable
+ Boolean getDeleteBatchEnabled() {
+ return deleteBatchEnabled;
+ }
+
boolean hasAnyOverride() {
return region != null
|| endpoint != null
@@ -144,7 +151,8 @@ final class S3BucketConfig {
|| assumeRoleExternalId != null
|| assumeRoleSessionName != null
|| assumeRoleSessionDurationSeconds != null
- || credentialsProvider != null;
+ || credentialsProvider != null
+ || deleteBatchEnabled != null;
}
@Override
@@ -169,7 +177,8 @@ final class S3BucketConfig {
&& Objects.equals(assumeRoleSessionName,
that.assumeRoleSessionName)
&& Objects.equals(
assumeRoleSessionDurationSeconds,
that.assumeRoleSessionDurationSeconds)
- && Objects.equals(credentialsProvider,
that.credentialsProvider);
+ && Objects.equals(credentialsProvider,
that.credentialsProvider)
+ && Objects.equals(deleteBatchEnabled, that.deleteBatchEnabled);
}
@Override
@@ -187,7 +196,8 @@ final class S3BucketConfig {
assumeRoleExternalId,
assumeRoleSessionName,
assumeRoleSessionDurationSeconds,
- credentialsProvider);
+ credentialsProvider,
+ deleteBatchEnabled);
}
@Override
@@ -228,6 +238,9 @@ final class S3BucketConfig {
if (credentialsProvider != null) {
sb.append(",
credentialsProvider='").append(credentialsProvider).append("'");
}
+ if (deleteBatchEnabled != null) {
+ sb.append(", deleteBatchEnabled=").append(deleteBatchEnabled);
+ }
sb.append('}');
return sb.toString();
}
@@ -253,6 +266,7 @@ final class S3BucketConfig {
@Nullable private String assumeRoleSessionName;
@Nullable private Integer assumeRoleSessionDurationSeconds;
@Nullable private String credentialsProvider;
+ @Nullable private Boolean deleteBatchEnabled;
private Builder(String bucketName) {
this.bucketName = bucketName;
@@ -322,6 +336,11 @@ final class S3BucketConfig {
return this;
}
+ Builder deleteBatchEnabled(boolean deleteBatchEnabled) {
+ this.deleteBatchEnabled = deleteBatchEnabled;
+ return this;
+ }
+
S3BucketConfig build() {
boolean hasAccessKey =
!StringUtils.isNullOrWhitespaceOnly(accessKey);
boolean hasSecretKey =
!StringUtils.isNullOrWhitespaceOnly(secretKey);
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/BucketConfigProviderTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/BucketConfigProviderTest.java
index 97921ff3175..be39aed2764 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/BucketConfigProviderTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/BucketConfigProviderTest.java
@@ -29,7 +29,7 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link BucketConfigProvider}. */
class BucketConfigProviderTest {
- /** One test exercises all 11 known properties on a single bucket. */
+ /** One test exercises all 12 known properties on a single bucket. */
@Test
void testParsesAllKnownPropertiesForSingleBucket() {
Configuration config = new Configuration();
@@ -52,6 +52,7 @@ class BucketConfigProviderTest {
config.setString(
"s3.bucket.my-bucket.aws.credentials.provider",
"software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider");
+ config.setString("s3.bucket.my-bucket.delete.batch.enabled", "false");
BucketConfigProvider provider = new BucketConfigProvider(config);
@@ -73,6 +74,7 @@ class BucketConfigProviderTest {
assertThat(bucket.getAssumeRoleSessionDurationSeconds()).isEqualTo(7200);
assertThat(bucket.getCredentialsProvider())
.isEqualTo("software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider");
+ assertThat(bucket.getDeleteBatchEnabled()).isFalse();
}
@Test
@@ -218,6 +220,17 @@ class BucketConfigProviderTest {
.hasMessageContaining("Invalid path-style-access");
}
+ @Test
+ void testInvalidDeleteBatchEnabledThrowsException() {
+ Configuration config = new Configuration();
+ config.setString("s3.bucket.my-bucket.delete.batch.enabled", "treu");
+ config.setString("s3.bucket.my-bucket.region", "us-east-1");
+
+ assertThatThrownBy(() -> new BucketConfigProvider(config))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("Invalid delete.batch.enabled");
+ }
+
@Test
void testUnrecognizedBucketPropertyIsIgnoredWithoutThrow() {
Configuration config = new Configuration();
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
index 5ed2edd1614..2d927d4fde7 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
@@ -218,6 +218,29 @@ class NativeS3FileSystemFactoryTest {
.isEqualTo(256 * 1024);
}
+ // --- Delete batching ---
+
+ @Test
+ void testDeleteBatchEnabledDefaultIsTrue() throws Exception {
+ assertThat(createFs(baseConfig()).isDeleteBatchEnabled()).isTrue();
+ }
+
+ @Test
+ void testDeleteBatchEnabledExplicitlyDisabled() throws Exception {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.DELETE_BATCH_ENABLED, false);
+ assertThat(createFs(config).isDeleteBatchEnabled()).isFalse();
+ }
+
+ @Test
+ void testDeleteBatchEnabledBucketOverridesGlobal() throws Exception {
+ // Global: true; bucket: false → bucket wins
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.DELETE_BATCH_ENABLED, true);
+ config.setString("s3.bucket.test-bucket.delete.batch.enabled",
"false");
+ assertThat(createFs(config).isDeleteBatchEnabled()).isFalse();
+ }
+
// --- Max connections ---
@Test
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java
index 5b10a9319d2..ac708c69400 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java
@@ -32,6 +32,8 @@ import org.apache.flink.core.testutils.TestContainerExtension;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.net.URI;
import java.nio.charset.StandardCharsets;
@@ -109,6 +111,31 @@ class NativeS3FileSystemITCase {
.doesNotThrowAnyException();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testRecursiveDeleteManyFiles(boolean batchingEnabled) throws
Exception {
+ final Configuration config = new Configuration();
+ container().setS3ConfigOptions(config);
+ config.set(NativeS3FileSystemFactory.DELETE_BATCH_ENABLED,
batchingEnabled);
+
+ final NativeS3FileSystemFactory factory = new
NativeS3FileSystemFactory();
+ factory.configure(config);
+ final FileSystem targetFs = factory.create(URI.create(bucketUri +
"/"));
+
+ final String dir = "bulk-delete-" + UUID.randomUUID();
+ final int numFiles = 25;
+ for (int i = 0; i < numFiles; i++) {
+ write(
+ targetFs,
+ path(dir + "/file-" + i + ".txt"),
+ ("data-" + i).getBytes(StandardCharsets.UTF_8));
+ }
+
+ assertThat(targetFs.listStatus(path(dir))).hasSize(numFiles);
+ assertThat(targetFs.delete(path(dir), true)).isTrue();
+ assertThat(targetFs.exists(path(dir))).isFalse();
+ }
+
@Test
void testRecoverableWriterMultipartCommit() throws Exception {
final Path file = path("recoverable-" + UUID.randomUUID() + ".bin");
@@ -132,7 +159,11 @@ class NativeS3FileSystemITCase {
}
private static void write(Path path, byte[] data) throws Exception {
- try (FSDataOutputStream out = fs.create(path,
FileSystem.WriteMode.OVERWRITE)) {
+ write(fs, path, data);
+ }
+
+ private static void write(FileSystem targetFs, Path path, byte[] data)
throws Exception {
+ try (FSDataOutputStream out = targetFs.create(path,
FileSystem.WriteMode.OVERWRITE)) {
out.write(data);
}
}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3RecursiveDeleteTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3RecursiveDeleteTest.java
new file mode 100644
index 00000000000..15fbf53748c
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3RecursiveDeleteTest.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3native;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectResponse;
+import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
+import software.amazon.awssdk.services.s3.model.S3Error;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link NativeS3RecursiveDelete}. */
+class NativeS3RecursiveDeleteTest {
+
+ private static final String BUCKET = "test-bucket";
+
+ @Test
+ void batchingEnabledIssuesOneDeleteObjectsCallAndNoDeleteObjectCalls()
throws Exception {
+ try (RecordingS3Client client = new RecordingS3Client(keys(5), 1000)) {
+ new NativeS3RecursiveDelete(client, BUCKET, "dir", true).execute();
+
+ assertThat(client.deleteObjectsRequests).hasSize(1);
+ assertThat(client.deleteObjectsRequests.get(0).delete().objects())
+ .extracting(ObjectIdentifier::key)
+ .containsExactlyElementsOf(keys(5));
+ assertThat(client.deleteObjectRequests).isEmpty();
+ }
+ }
+
+ @Test
+ void
batchingDisabledIssuesOneDeleteObjectCallPerKeyAndNoDeleteObjectsCalls() throws
Exception {
+ try (RecordingS3Client client = new RecordingS3Client(keys(5), 1000)) {
+ new NativeS3RecursiveDelete(client, BUCKET, "dir",
false).execute();
+
+ assertThat(client.deleteObjectRequests)
+ .extracting(DeleteObjectRequest::key)
+ .containsExactlyElementsOf(keys(5));
+ assertThat(client.deleteObjectsRequests).isEmpty();
+ }
+ }
+
+ @Test
+ void moreThanPageSizeKeysAreSplitAcrossMultipleDeleteObjectsCalls() throws
Exception {
+ // Small page size to make pagination and per-page batching observable
without
+ // constructing 1000+ keys.
+ try (RecordingS3Client client = new RecordingS3Client(keys(7), 3)) {
+ new NativeS3RecursiveDelete(client, BUCKET, "dir", true).execute();
+
+ // 7 keys at 3 per listing page -> 3 DeleteObjects calls (3, 3, 1
keys), each within
+ // the page size, and never holding more than one page of keys at
a time.
+ assertThat(client.deleteObjectsRequests).hasSize(3);
+ assertThat(
+ client.deleteObjectsRequests.stream()
+ .flatMap(r ->
r.delete().objects().stream())
+ .map(ObjectIdentifier::key)
+ .collect(Collectors.toList()))
+ .containsExactlyElementsOf(keys(7));
+ assertThat(client.deleteObjectsRequests)
+ .allSatisfy(r ->
assertThat(r.delete().objects()).hasSizeLessThanOrEqualTo(3));
+ }
+ }
+
+ @Test
+ void partialBatchDeleteFailureThrowsIOException() throws Exception {
+ try (RecordingS3Client client = new RecordingS3Client(keys(3), 1000)) {
+ client.failKeyWith(
+ keys(3).get(1),
S3Error.builder().code("AccessDenied").message("nope").build());
+
+ assertThatThrownBy(
+ () ->
+ new NativeS3RecursiveDelete(client,
BUCKET, "dir", true)
+ .execute())
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining(keys(3).get(1))
+ .hasMessageContaining("AccessDenied");
+ }
+ }
+
+ private static List<String> keys(int count) {
+ return IntStream.range(0, count)
+ .mapToObj(i -> "dir/file-" + i)
+ .collect(Collectors.toList());
+ }
+
+ /** Records every {@code list}/{@code delete} request issued against a
fixed key set. */
+ private static final class RecordingS3Client implements S3Client {
+ private final List<String> allKeys;
+ private final int pageSize;
+ private final Map<String, S3Error> failuresByKey = new HashMap<>();
+
+ final List<DeleteObjectsRequest> deleteObjectsRequests = new
ArrayList<>();
+ final List<DeleteObjectRequest> deleteObjectRequests = new
ArrayList<>();
+
+ RecordingS3Client(List<String> allKeys, int pageSize) {
+ this.allKeys = allKeys;
+ this.pageSize = pageSize;
+ }
+
+ void failKeyWith(String key, S3Error error) {
+ failuresByKey.put(key, error);
+ }
+
+ @Override
+ public ListObjectsV2Response listObjectsV2(ListObjectsV2Request
request) {
+ int start =
+ request.continuationToken() == null
+ ? 0
+ : Integer.parseInt(request.continuationToken());
+ int end = Math.min(start + pageSize, allKeys.size());
+
+ List<S3Object> page =
+ allKeys.subList(start, end).stream()
+ .map(k -> S3Object.builder().key(k).build())
+ .collect(Collectors.toList());
+
+ ListObjectsV2Response.Builder builder =
ListObjectsV2Response.builder().contents(page);
+ if (end < allKeys.size()) {
+
builder.nextContinuationToken(String.valueOf(end)).isTruncated(true);
+ }
+ return builder.build();
+ }
+
+ @Override
+ public DeleteObjectsResponse deleteObjects(DeleteObjectsRequest
request) {
+ deleteObjectsRequests.add(request);
+ List<S3Error> errors = new ArrayList<>();
+ for (ObjectIdentifier id : request.delete().objects()) {
+ S3Error error = failuresByKey.get(id.key());
+ if (error != null) {
+ errors.add(error.toBuilder().key(id.key()).build());
+ }
+ }
+ return DeleteObjectsResponse.builder().errors(errors).build();
+ }
+
+ @Override
+ public DeleteObjectResponse deleteObject(DeleteObjectRequest request) {
+ deleteObjectRequests.add(request);
+ return DeleteObjectResponse.builder().build();
+ }
+
+ @Override
+ public String serviceName() {
+ return "s3";
+ }
+
+ @Override
+ public void close() {}
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/resources/log4j2-test.properties
b/flink-filesystems/flink-s3-fs-native/src/test/resources/log4j2-test.properties
new file mode 100644
index 00000000000..835c2ec9a3d
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/resources/log4j2-test.properties
@@ -0,0 +1,28 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
