This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi 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 b0c5254ce36 [FLINK-39110][s3] Add CRT support in native-s3-fs
b0c5254ce36 is described below
commit b0c5254ce36d3d61e55600a9d9356fe0b861cc6c
Author: Samrat <[email protected]>
AuthorDate: Thu Jul 30 17:09:21 2026 +0530
[FLINK-39110][s3] Add CRT support in native-s3-fs
---
flink-filesystems/flink-s3-fs-native/README.md | 83 ++++-
flink-filesystems/flink-s3-fs-native/pom.xml | 37 +++
.../flink/fs/s3native/NativeS3BulkCopyHelper.java | 369 +++++++++++++++++----
.../flink/fs/s3native/NativeS3FileIoUtils.java | 86 +++++
.../flink/fs/s3native/NativeS3FileSystem.java | 8 +-
.../fs/s3native/NativeS3FileSystemFactory.java | 166 ++++++++-
.../flink/fs/s3native/NativeS3OutputStream.java | 4 +-
.../apache/flink/fs/s3native/S3ClientProvider.java | 353 +++++++++++++++++---
.../s3native/writer/NativeS3ObjectOperations.java | 58 +++-
.../NativeS3RecoverableFsDataOutputStream.java | 30 +-
.../src/main/resources/META-INF/NOTICE | 1 +
.../fs/s3native/NativeS3BulkCopyHelperTest.java | 326 +++++++++++++++++-
.../flink/fs/s3native/NativeS3FileIoUtilsTest.java | 90 +++++
.../fs/s3native/NativeS3FileSystemFactoryTest.java | 169 +++++++++-
.../flink/fs/s3native/S3ClientProviderTest.java | 89 +++++
.../flink-s3-fs-native/tools/download-crt-jars.sh | 134 ++++++++
16 files changed, 1847 insertions(+), 156 deletions(-)
diff --git a/flink-filesystems/flink-s3-fs-native/README.md
b/flink-filesystems/flink-s3-fs-native/README.md
index 292f87cd25a..d8be409626b 100644
--- a/flink-filesystems/flink-s3-fs-native/README.md
+++ b/flink-filesystems/flink-s3-fs-native/README.md
@@ -71,9 +71,10 @@ input.sinkTo(FileSink.forRowFormat(new
Path("s3://my-bucket/output"),
| s3.upload.max.concurrent.uploads | CPU cores | Maximum concurrent part
uploads per stream |
| s3.entropy.key | (none) | Key for entropy injection in paths |
| s3.entropy.length | 4 | Length of entropy string |
-| s3.bulk-copy.enabled | true | Enable bulk copy operations using
S3TransferManager |
+| s3.bulk-copy.enabled | true | Enable bulk copy operations for S3-to-local
downloads |
| s3.bulk-copy.max-concurrent | 16 | Maximum number of concurrent copy
operations |
-| s3.connection.max | 50 | Maximum HTTP connections in the S3 client
connection pool. Applies to both sync (Apache HTTP) and async (Netty) clients.
Must be ≥ `s3.bulk-copy.max-concurrent` |
+| s3.bulk-copy.download-buffer-size | 262144 (256KB) | Buffer size for writing
bulk-copy downloads to local disk. Bounds the JDK's cached temporary direct
buffers, preventing direct-memory OutOfMemoryError during large RocksDB state
restores |
+| 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) |
@@ -357,6 +358,84 @@ When enabled, file uploads automatically use
TransferManager for:
- Better utilization of available bandwidth
- Lower heap requirements for write operations
+## AWS Common Runtime (CRT) Support
+
+The filesystem optionally supports the [AWS Common Runtime
(CRT)](https://github.com/awslabs/aws-crt-java) HTTP transport
+for higher throughput on large S3 workloads.
+
+When enabled, the CRT transport replaces:
+- **Sync client**: Apache HTTP Client → `AwsCrtHttpClient`
+- **Async client**: Netty NIO → `S3AsyncClient.crtBuilder()` (with built-in
multipart acceleration)
+
+### Prerequisites
+
+The `aws-crt` artifact contains JNI-linked native libraries whose C-side
`FindClass` paths are
+hardcoded, making Maven shade relocation incompatible. Therefore **the
`aws-crt` JAR is not
+bundled** in the fat JAR and must be placed manually.
+
+### Setup
+
+1. From the module directory, run the helper script to download the `aws-crt`
+ JAR (auto-resolves the compatible version for the AWS SDK version this
+ module was built against):
+
+ ```bash
+ ./tools/download-crt-jars.sh
+ ```
+
+ This places the JAR in `./crt-jars/`. Pass a different directory as the
+ first argument if needed. Requires `mvn` on `PATH`.
+
+2. Copy the JAR into the Flink plugin directory alongside
`flink-s3-fs-native.jar`:
+
+ ```bash
+ cp crt-jars/aws-crt-*.jar $FLINK_HOME/plugins/s3-fs-native/
+ ```
+
+ > **Note:** `aws-crt-client` does **not** need to be downloaded or placed
+ > separately — it is bundled (shaded) directly inside
`flink-s3-fs-native.jar`
+ > at build time. Only `aws-crt` (the JNI native lib) must be placed manually
+ > because its C-side `FindClass` paths are hardcoded and incompatible with
+ > Maven shade relocation.
+
+3. Enable CRT in your Flink configuration (`conf/config.yaml`):
+
+ ```yaml
+ s3.crt.enabled: true
+ ```
+
+If the `aws-crt` JAR is missing when `s3.crt.enabled: true`, the filesystem
+fails fast at startup with an `IllegalStateException` pointing back to this
+setup procedure.
+
+#### Manual download (alternative)
+
+If `mvn` is unavailable, fetch the JAR by hand from Maven Central:
+
+`aws-crt-<version>.jar` (groupId: `software.amazon.awssdk.crt`) — the version
is
+declared as the `<dependency>` on `software.amazon.awssdk.crt:aws-crt` inside
+`aws-crt-client-<version>.pom` on Maven Central. The `aws-crt` artifact uses an
+independent versioning scheme (e.g. `0.45.x`) that does **not** track the AWS
SDK
+version.
+
+`aws-crt-client` does **not** need to be downloaded — it is bundled inside the
fat JAR.
+
+### CRT Configuration Options
+
+| Key | Default | Description |
+|-----|---------|-------------|
+| s3.crt.enabled | false | Enable CRT HTTP transport for both sync and async
S3 clients |
+| s3.crt.target-throughput-gbps | (none) | Soft target throughput in Gbps for
the CRT async client. Hint, not a hard cap — actual throughput may exceed the
configured value. When unset, the AWS CRT runtime applies its own internal
default; set this only to override it (e.g. to match the network bandwidth
available to a single TaskManager). |
+| s3.crt.max-native-memory-limit | (none) | Maximum native memory the CRT
async client may use. When unset, the AWS CRT runtime applies its own internal
limit. |
+| s3.crt.read-buffer-size | (none) | Read buffer size for the CRT sync and
async clients. Decoupled from `s3.read.buffer.size` so lowering the streaming
read buffer does not shrink CRT's native transfer buffers. When unset, the AWS
CRT runtime applies its own (larger) default. |
+| s3.crt.max-concurrency | 256 | Max concurrent in-flight requests for the CRT
sync and async clients. Decoupled from `s3.connection.max` because CRT fans one
logical transfer into many part-sized requests; reusing the smaller
connection-pool size causes "failed to acquire a connection" timeouts under
parallel checkpoint upload/restore. |
+
+The CRT read buffer is controlled independently via `s3.crt.read-buffer-size`;
when unset, the CRT runtime keeps its own default rather than inheriting
`s3.read.buffer.size`.
+
+> **Note on options silently ignored under CRT:**
+> - `s3.socket.timeout` — `AwsCrtHttpClient` has no socket-level read timeout;
the CRT runtime uses `ConnectionHealthConfiguration` for stalled-read detection
instead.
+> - `s3.chunked-encoding.enabled` — `S3CrtAsyncClientBuilder` manages wire
encoding internally and exposes no equivalent setter.
+
## Checkpointing
Configure checkpoint storage in `conf/config.yaml`:
diff --git a/flink-filesystems/flink-s3-fs-native/pom.xml
b/flink-filesystems/flink-s3-fs-native/pom.xml
index cdc644d58af..b0403112df0 100644
--- a/flink-filesystems/flink-s3-fs-native/pom.xml
+++ b/flink-filesystems/flink-s3-fs-native/pom.xml
@@ -32,6 +32,10 @@ under the License.
<properties>
<fs.s3.aws.sdk.version>2.44.4</fs.s3.aws.sdk.version>
+ <!-- aws-crt uses a separate versioning scheme from the AWS
SDK; this version was
+ validated against aws-crt-client ${fs.s3.aws.sdk.version}.
Update when bumping
+ the SDK version (check aws-crt-client-<version>.pom →
awscrt.version). -->
+ <fs.s3.aws.crt.version>0.45.1</fs.s3.aws.crt.version>
<japicmp.skip>true</japicmp.skip>
<surefire.module.config>--add-opens=java.base/java.util=ALL-UNNAMED</surefire.module.config>
</properties>
@@ -89,6 +93,22 @@ under the License.
<optional>${flink.markBundledAsOptional}</optional>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>aws-crt-client</artifactId>
+ <version>${fs.s3.aws.sdk.version}</version>
+ <optional>${flink.markBundledAsOptional}</optional>
+ <exclusions>
+ <!-- aws-crt contains JNI native libs with
hardcoded FindClass paths;
+ it cannot be bundled or shaded. It must be
placed alongside the
+ fat JAR in the plugin directory with its
original class names. -->
+ <exclusion>
+
<groupId>software.amazon.awssdk.crt</groupId>
+ <artifactId>aws-crt</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
@@ -135,6 +155,15 @@ under the License.
<artifactId>testcontainers-junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
+
+ <!-- aws-crt is excluded from production fat-JAR (JNI shading
incompatibility) but
+ must be on the test classpath for CRT-enabled unit tests
to instantiate clients. -->
+ <dependency>
+ <groupId>software.amazon.awssdk.crt</groupId>
+ <artifactId>aws-crt</artifactId>
+ <version>${fs.s3.aws.crt.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
@@ -164,6 +193,14 @@ under the License.
<relocation>
<pattern>software.amazon.awssdk</pattern>
<shadedPattern>org.apache.flink.fs.s3native.shaded.software.amazon.awssdk</shadedPattern>
+
<excludes>
+
<!-- aws-crt JNI bridge classes must keep their original names.
+
The C-side FindClass paths are hardcoded in native code and
+
cannot be relocated. aws-crt-client (pure Java) is bundled
+
and shaded, but its calls into aws-crt must resolve against
+
the external aws-crt JAR using the original package names. -->
+
<exclude>software.amazon.awssdk.crt.**</exclude>
+
</excludes>
</relocation>
<relocation>
<pattern>org.apache.http</pattern>
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java
index 5fde18815d3..e077cb0635f 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java
@@ -22,26 +22,40 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.fs.ICloseableRegistry;
import org.apache.flink.core.fs.PathsCopyingFileSystem;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.transfer.s3.S3TransferManager;
-import software.amazon.awssdk.transfer.s3.model.CompletedCopy;
-import software.amazon.awssdk.transfer.s3.model.DownloadFileRequest;
-import software.amazon.awssdk.transfer.s3.model.FileDownload;
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.flink.util.Preconditions.checkArgument;
/**
- * Helper class for performing bulk S3 to local file system copies using
S3TransferManager.
+ * Helper class for performing bulk S3 to local file system copies using the
S3 async client.
*
* <p><b>Concurrency Model:</b> Uses batch-based concurrency control with
{@code
* maxConcurrentCopies} to limit parallel downloads. The effective concurrency
is clamped to the
@@ -51,18 +65,19 @@ import static
org.apache.flink.util.Preconditions.checkArgument;
* bounded executor) to allow continuous submission of new downloads as slots
become available,
* which would provide better throughput by avoiding the "slowest task in
batch" bottleneck.
*
- * <p><b>Retry Handling:</b> Relies on the S3TransferManager's built-in retry
mechanism for
- * transient failures. If a download fails after retries:
+ * <p><b>Retry Handling:</b> Relies on the S3 async client's built-in retry
mechanism for transient
+ * failures. If a download fails after retries:
*
* <ul>
* <li>The entire bulk copy operation fails with an IOException
* <li>Successfully downloaded files are NOT cleaned up (they remain on disk)
- * <li>Partial downloads may leave incomplete files that should be cleaned
up by the caller
+ * <li>The failed file's temporary download is deleted before the method
returns
* </ul>
*
- * <p><b>Cleanup:</b> No automatic cleanup is performed on failure. Callers
are responsible for
- * cleaning up destination files if the bulk copy fails. Consider wrapping in
a try-finally or using
- * a temp directory that can be deleted on failure.
+ * <p><b>Cleanup:</b> Each file is downloaded into a temporary file in the
destination directory and
+ * atomically moved into place after the stream has been fully copied.
Cancellation through the
+ * provided {@link ICloseableRegistry} aborts active S3 response streams,
cancels pending futures,
+ * stops the worker pool, and deletes incomplete temporary files.
*
* <p><b>TODO:</b> Consider extracting URI parsing logic to a shared
S3UriUtils utility class to
* consolidate S3 URI handling across the codebase.
@@ -72,24 +87,32 @@ class NativeS3BulkCopyHelper {
private static final Logger LOG =
LoggerFactory.getLogger(NativeS3BulkCopyHelper.class);
- private final S3TransferManager transferManager;
+ private final S3AsyncClient asyncClient;
private final int maxConcurrentCopies;
private final int maxConnections;
+ private final int downloadBufferSize;
/**
* Creates a new bulk copy helper.
*
- * @param transferManager the S3 transfer manager for async downloads
+ * @param asyncClient the S3 async client used for downloads
* @param maxConcurrentCopies the requested maximum number of concurrent
copy operations
* @param maxConnections the HTTP connection pool size; if {@code
maxConcurrentCopies} exceeds
* this value, it is clamped down to prevent connection pool exhaustion
+ * @param downloadBufferSize the buffer size in bytes used to write each
downloaded file to the
+ * local filesystem; bounds the temporary direct buffers the JDK
caches for channel writes
*/
NativeS3BulkCopyHelper(
- S3TransferManager transferManager, int maxConcurrentCopies, int
maxConnections) {
+ S3AsyncClient asyncClient,
+ int maxConcurrentCopies,
+ int maxConnections,
+ int downloadBufferSize) {
checkArgument(maxConcurrentCopies > 0, "maxConcurrentCopies must be
positive");
checkArgument(maxConnections > 0, "maxConnections must be positive");
- this.transferManager = transferManager;
+ checkArgument(downloadBufferSize > 0, "downloadBufferSize must be
positive");
+ this.asyncClient = asyncClient;
this.maxConnections = maxConnections;
+ this.downloadBufferSize = downloadBufferSize;
if (maxConcurrentCopies > maxConnections) {
LOG.warn(
"{} ({}) exceeds {} ({}). "
@@ -109,16 +132,20 @@ class NativeS3BulkCopyHelper {
return maxConcurrentCopies;
}
+ @VisibleForTesting
+ int getDownloadBufferSize() {
+ return downloadBufferSize;
+ }
+
/**
* Copies files from S3 to local filesystem in batches.
*
- * <p><b>Error Handling:</b> If an unsupported URI scheme is encountered,
all already-started
- * copy operations are awaited to completion before throwing the
exception. This ensures that no
- * background copy tasks are left running when the method returns,
allowing the caller to safely
- * manage cleanup and resource lifecycle.
+ * <p><b>Error Handling:</b> If an unsupported URI scheme or copy failure
is encountered,
+ * already-started copy operations are cancelled before throwing the
exception. Successfully
+ * completed destination files are left in place; incomplete temporary
files are deleted.
*
* @param requests List of copy requests (source S3 path to destination
local path)
- * @param closeableRegistry Registry for cleanup (currently unused,
reserved for future use)
+ * @param closeableRegistry Registry for cancelling in-flight copies
during task cancellation
* @throws IOException if any copy operation fails or if an unsupported
URI scheme is
* encountered
*/
@@ -133,24 +160,40 @@ class NativeS3BulkCopyHelper {
int totalFiles = requests.size();
int totalBatches = (totalFiles + maxConcurrentCopies - 1) /
maxConcurrentCopies;
LOG.info(
- "Starting bulk copy of {} files using S3TransferManager "
- + "(batch size: {}, total batches: {})",
+ "Starting bulk copy of {} files (batch size: {}, total
batches: {})",
totalFiles,
maxConcurrentCopies,
totalBatches);
- List<CompletableFuture<CompletedCopy>> copyFutures = new ArrayList<>();
+ ExecutorService downloadPool =
+ Executors.newFixedThreadPool(
+ maxConcurrentCopies,
+ new ExecutorThreadFactory(
+ "s3-native-bulk-copy",
+ (thread, error) ->
+ LOG.error(
+ "Uncaught exception in S3
bulk-copy worker {}",
+ thread.getName(),
+ error)));
+ BulkCopyCancellation cancellation = new
BulkCopyCancellation(downloadPool);
+ ICloseableRegistry registry =
+ closeableRegistry == null ? ICloseableRegistry.NO_OP :
closeableRegistry;
+ List<CompletableFuture<Void>> copyFutures = new ArrayList<>();
int batchNumber = 0;
- try {
+ try (Closeable ignored =
registry.registerCloseableTemporarily(cancellation)) {
for (int i = 0; i < requests.size(); i++) {
PathsCopyingFileSystem.CopyRequest request = requests.get(i);
String sourceUri = request.getSource().toUri().toString();
- if (sourceUri.startsWith("s3://") ||
sourceUri.startsWith("s3a://")) {
- copyFutures.add(copyS3ToLocal(request));
+ if (isSupportedS3Scheme(request.getSource())
+ && isSupportedLocalScheme(request.getDestination())) {
+ copyFutures.add(copyS3ToLocal(request, downloadPool,
cancellation));
} else {
throw new UnsupportedOperationException(
- "Only S3 to local copies are currently supported:
" + sourceUri);
+ "Only S3 to local copies are currently supported: "
+ + sourceUri
+ + " -> "
+ + request.getDestination());
}
if (copyFutures.size() >= maxConcurrentCopies || i ==
requests.size() - 1) {
@@ -166,69 +209,168 @@ class NativeS3BulkCopyHelper {
}
LOG.info("Completed bulk copy of {} files", totalFiles);
- } catch (Exception e) {
- if (!copyFutures.isEmpty()) {
- LOG.warn(
- "Error during bulk copy, waiting for {} in-flight
operations to complete",
- copyFutures.size());
- try {
- waitForCopies(copyFutures);
- } catch (IOException waitError) {
- LOG.warn(
- "Error waiting for in-flight copy operations: {}",
- waitError.getMessage());
- e.addSuppressed(waitError);
- }
- }
- if (e instanceof IOException) {
- throw (IOException) e;
- } else {
- throw new IOException(e);
- }
+ } catch (Throwable e) {
+ cancellation.close();
+ ExceptionUtils.rethrowIOException(e);
+ } finally {
+ downloadPool.shutdownNow();
}
}
/**
* Initiates an async S3 to local file copy.
*
- * @param request The copy request containing source S3 path and
destination local path
- * @return A CompletableFuture that completes when the download finishes
+ * <p>The object body is streamed to disk through a bounded {@code byte[]}
buffer (see {@code
+ * s3.bulk-copy.download-buffer-size}) rather than the SDK's {@code
AsynchronousFileChannel}
+ * based {@code toFile} transformer, which caches an unbounded per-thread
temporary direct
+ * buffer sized to each write and can exhaust direct memory during large
restores.
+ *
+ * @param request the copy request containing source S3 path and
destination local path
+ * @param downloadPool the executor on which the blocking stream copy runs
+ * @return a CompletableFuture that completes when the download finishes
* @throws IOException if the destination directory cannot be created
*/
- private CompletableFuture<CompletedCopy> copyS3ToLocal(
- PathsCopyingFileSystem.CopyRequest request) throws IOException {
+ private CompletableFuture<Void> copyS3ToLocal(
+ PathsCopyingFileSystem.CopyRequest request,
+ ExecutorService downloadPool,
+ BulkCopyCancellation cancellation)
+ throws IOException {
String sourceUri = request.getSource().toUri().toString();
String bucket = extractBucket(sourceUri);
String key = extractKey(sourceUri);
- File destFile = new File(request.getDestination().getPath());
+ Path destination = new
File(request.getDestination().getPath()).toPath().toAbsolutePath();
- Files.createDirectories(destFile.getParentFile().toPath());
+ Path parent = destination.getParent();
+ if (parent != null) {
+ Files.createDirectories(parent);
+ }
+ Path tempDestination =
NativeS3FileIoUtils.createTemporaryDownloadFile(parent, destination);
- DownloadFileRequest downloadRequest =
- DownloadFileRequest.builder()
- .getObjectRequest(req -> req.bucket(bucket).key(key))
- .destination(destFile.toPath())
- .build();
+ GetObjectRequest getObjectRequest =
+ GetObjectRequest.builder().bucket(bucket).key(key).build();
- FileDownload download = transferManager.downloadFile(downloadRequest);
+ CompletableFuture<ResponseInputStream<GetObjectResponse>>
responseFuture;
+ try {
+ responseFuture =
+ asyncClient.getObject(
+ getObjectRequest,
AsyncResponseTransformer.toBlockingInputStream());
+ } catch (RuntimeException | Error e) {
+ IOUtils.deleteFileQuietly(tempDestination);
+ throw e;
+ }
+ cancellation.registerFuture(responseFuture);
+ CompletableFuture<Void> copyFuture = new CompletableFuture<>();
+ cancellation.registerFuture(copyFuture);
+ responseFuture.whenComplete(
+ (responseStream, error) -> {
+ cancellation.unregisterFuture(responseFuture);
+ if (error != null) {
+ IOUtils.deleteFileQuietly(tempDestination);
+ copyFuture.completeExceptionally(error);
+ return;
+ }
+ if (responseStream == null) {
+ IOUtils.deleteFileQuietly(tempDestination);
+ copyFuture.completeExceptionally(
+ new IOException(
+ "S3 getObject completed without a
response stream"));
+ return;
+ }
+ submitDownload(
+ responseStream,
+ tempDestination,
+ destination,
+ sourceUri,
+ downloadPool,
+ cancellation,
+ copyFuture);
+ });
+ copyFuture.whenComplete(
+ (ignored, error) -> {
+ cancellation.unregisterFuture(copyFuture);
+ if (copyFuture.isCancelled()) {
+ responseFuture.cancel(true);
+ }
+ if (error != null) {
+ IOUtils.deleteFileQuietly(tempDestination);
+ }
+ });
+ return copyFuture;
+ }
- return download.completionFuture()
- .thenApply(
- completed -> {
- LOG.debug("Successfully copied {} to {}",
sourceUri, destFile);
- return null;
- });
+ private void submitDownload(
+ ResponseInputStream<GetObjectResponse> responseStream,
+ Path tempDestination,
+ Path destination,
+ String sourceUri,
+ ExecutorService downloadPool,
+ BulkCopyCancellation cancellation,
+ CompletableFuture<Void> copyFuture) {
+ if (!cancellation.registerStream(responseStream)) {
+ abortAndClose(responseStream);
+ IOUtils.deleteFileQuietly(tempDestination);
+ copyFuture.completeExceptionally(new CancellationException("Bulk
copy was cancelled"));
+ return;
+ }
+ try {
+ downloadPool.execute(
+ () -> {
+ boolean success = false;
+ try {
+ NativeS3FileIoUtils.copyStream(
+ responseStream, tempDestination,
downloadBufferSize);
+ NativeS3FileIoUtils.moveFile(tempDestination,
destination);
+ success = true;
+ LOG.debug("Successfully copied {} to {}",
sourceUri, destination);
+ copyFuture.complete(null);
+ } catch (Throwable t) {
+ copyFuture.completeExceptionally(t);
+ } finally {
+ cancellation.unregisterStream(responseStream);
+ // Close on success (stream fully read, connection
reusable); abort on
+ // failure to drop the connection immediately
instead of draining a
+ // large partially-read body.
+ if (success) {
+ closeQuietly(responseStream);
+ } else {
+ abortAndClose(responseStream);
+ }
+ IOUtils.deleteFileQuietly(tempDestination);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ cancellation.unregisterStream(responseStream);
+ abortAndClose(responseStream);
+ IOUtils.deleteFileQuietly(tempDestination);
+ copyFuture.completeExceptionally(e);
+ }
}
- private void waitForCopies(List<CompletableFuture<CompletedCopy>> futures)
throws IOException {
+ private void waitForCopies(List<CompletableFuture<Void>> futures) throws
IOException {
try {
- CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).get();
+ // Fail fast: complete as soon as either all downloads finish
successfully or the first
+ // one fails, rather than waiting for every in-flight download to
run to completion. The
+ // outer copyFiles handler aborts the remaining streams and shuts
the pool down on the
+ // resulting exception.
+ CompletableFuture<Void> allDone =
+ CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0]));
+ CompletableFuture<Void> firstFailure = new CompletableFuture<>();
+ for (CompletableFuture<Void> future : futures) {
+ future.whenComplete(
+ (ignored, error) -> {
+ if (error != null) {
+ firstFailure.completeExceptionally(error);
+ }
+ });
+ }
+ CompletableFuture.anyOf(allDone, firstFailure).get();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Bulk copy interrupted", e);
} catch (ExecutionException e) {
Throwable cause = e.getCause();
+ ExceptionUtils.rethrowIfFatalError(cause);
if (isConnectionPoolExhausted(cause)) {
throw new IOException(
String.format(
@@ -246,6 +388,101 @@ class NativeS3BulkCopyHelper {
}
}
+ static boolean isSupportedS3Scheme(org.apache.flink.core.fs.Path path) {
+ String scheme = path.toUri().getScheme();
+ return "s3".equalsIgnoreCase(scheme) || "s3a".equalsIgnoreCase(scheme);
+ }
+
+ static boolean isSupportedLocalScheme(org.apache.flink.core.fs.Path path) {
+ String scheme = path.toUri().getScheme();
+ return scheme == null || "file".equalsIgnoreCase(scheme);
+ }
+
+ private static void abortAndClose(ResponseInputStream<GetObjectResponse>
stream) {
+ try {
+ stream.abort();
+ } catch (RuntimeException e) {
+ LOG.debug("Error aborting S3 response stream during bulk-copy
cancellation", e);
+ }
+ try {
+ stream.close();
+ } catch (IOException e) {
+ LOG.debug("Error closing S3 response stream during bulk-copy
cancellation", e);
+ }
+ }
+
+ private static void closeQuietly(ResponseInputStream<GetObjectResponse>
stream) {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ LOG.debug("Error closing S3 response stream after successful
bulk-copy download", e);
+ }
+ }
+
+ private static final class BulkCopyCancellation implements Closeable {
+ private static final long TERMINATION_TIMEOUT_SECONDS = 5L;
+
+ private final ExecutorService downloadPool;
+ private final Set<CompletableFuture<?>> futures =
+ Collections.newSetFromMap(new
java.util.concurrent.ConcurrentHashMap<>());
+ private final Set<ResponseInputStream<GetObjectResponse>>
activeStreams =
+ Collections.newSetFromMap(new
java.util.concurrent.ConcurrentHashMap<>());
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ private BulkCopyCancellation(ExecutorService downloadPool) {
+ this.downloadPool = downloadPool;
+ }
+
+ private void registerFuture(CompletableFuture<?> future) {
+ if (closed.get()) {
+ future.cancel(true);
+ return;
+ }
+ futures.add(future);
+ if (closed.get() && futures.remove(future)) {
+ future.cancel(true);
+ }
+ }
+
+ private void unregisterFuture(CompletableFuture<?> future) {
+ futures.remove(future);
+ }
+
+ private boolean registerStream(ResponseInputStream<GetObjectResponse>
stream) {
+ if (closed.get()) {
+ return false;
+ }
+ activeStreams.add(stream);
+ if (closed.get() && activeStreams.remove(stream)) {
+ return false;
+ }
+ return true;
+ }
+
+ private void unregisterStream(ResponseInputStream<GetObjectResponse>
stream) {
+ activeStreams.remove(stream);
+ }
+
+ @Override
+ public void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+ futures.forEach(future -> future.cancel(true));
+ activeStreams.forEach(NativeS3BulkCopyHelper::abortAndClose);
+ downloadPool.shutdownNow();
+ try {
+ if
(!downloadPool.awaitTermination(TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))
{
+ LOG.warn(
+ "S3 bulk-copy worker pool did not terminate within
{} seconds",
+ TERMINATION_TIMEOUT_SECONDS);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
/**
* Checks whether a failure was caused by HTTP connection pool exhaustion.
*
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileIoUtils.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileIoUtils.java
new file mode 100644
index 00000000000..34f35ebc469
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileIoUtils.java
@@ -0,0 +1,86 @@
+/*
+ * 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.apache.flink.annotation.Internal;
+import org.apache.flink.util.IOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.AtomicMoveNotSupportedException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+
+/**
+ * Shared local-filesystem helpers for the S3 download paths (bulk copy and
single-object get).
+ *
+ * <p>Downloads stream the S3 object body to disk through a bounded heap
{@code byte[]} rather than
+ * the SDK's {@code AsynchronousFileChannel}-based {@code toFile} transformer,
which caches an
+ * unbounded per-thread temporary direct buffer sized to each write and can
exhaust direct memory
+ * during large restores. Keeping this logic in one place ensures both
download paths stay bounded.
+ */
+@Internal
+public final class NativeS3FileIoUtils {
+
+ private NativeS3FileIoUtils() {}
+
+ /**
+ * Copies all bytes from {@code in} to {@code destination}, overwriting
any existing file, using
+ * a fixed-size heap buffer. The caller retains ownership of {@code in}
and is responsible for
+ * closing (or aborting) it — this method only closes the destination
stream so the correct
+ * close-vs-abort decision can be made based on success or failure.
+ */
+ public static void copyStream(InputStream in, Path destination, int
bufferSize)
+ throws IOException {
+ try (OutputStream out = Files.newOutputStream(destination)) {
+ IOUtils.copyBytes(in, out, bufferSize, false);
+ }
+ }
+
+ /** Moves {@code source} onto {@code destination}, preferring an atomic
move when supported. */
+ public static void moveFile(Path source, Path destination) throws
IOException {
+ try {
+ Files.move(
+ source,
+ destination,
+ StandardCopyOption.REPLACE_EXISTING,
+ StandardCopyOption.ATOMIC_MOVE);
+ } catch (AtomicMoveNotSupportedException ignored) {
+ Files.move(source, destination,
StandardCopyOption.REPLACE_EXISTING);
+ }
+ }
+
+ /**
+ * Creates a temporary download file in {@code parent} whose name is
derived from {@code
+ * destination}. The prefix is padded to satisfy {@link
Files#createTempFile}'s minimum length.
+ */
+ public static Path createTemporaryDownloadFile(Path parent, Path
destination)
+ throws IOException {
+ String prefix =
+ destination.getFileName() == null
+ ? "s3-download"
+ : destination.getFileName().toString();
+ if (prefix.length() < 3) {
+ prefix = "s3-" + prefix;
+ }
+ return Files.createTempFile(parent, prefix, ".tmp");
+ }
+}
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 0ec9fb22624..efec0dcc185 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
@@ -512,7 +512,9 @@ class NativeS3FileSystem extends FileSystem
@Override
public boolean canCopyPaths(Path source, Path destination) {
- return bulkCopyHelper != null;
+ return bulkCopyHelper != null
+ && NativeS3BulkCopyHelper.isSupportedS3Scheme(source)
+ && NativeS3BulkCopyHelper.isSupportedLocalScheme(destination);
}
@Override
@@ -566,12 +568,12 @@ class NativeS3FileSystem extends FileSystem
"S3
client provider closed");
}
}))
- .orTimeout(fsCloseTimeout.toSeconds(),
TimeUnit.SECONDS)
+ .orTimeout(fsCloseTimeout.toMillis(),
TimeUnit.MILLISECONDS)
.whenComplete(
(result, error) -> {
if (error != null) {
LOG.error(
- "FileSystem close timed out
after {} for bucket: {}",
+ "FileSystem close did not
complete cleanly within {} for bucket: {}",
fsCloseTimeout,
bucketName,
error);
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 2d7fea8532e..cb4fd4788f5 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
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystemFactory;
import org.apache.flink.util.Preconditions;
@@ -144,7 +145,7 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
ConfigOptions.key("s3.bulk-copy.enabled")
.booleanType()
.defaultValue(true)
- .withDescription("Enable bulk copy operations using
S3TransferManager");
+ .withDescription("Enable bulk copy operations for
S3-to-local downloads");
public static final ConfigOption<Integer> MAX_CONNECTIONS =
ConfigOptions.key("s3.connection.max")
@@ -152,7 +153,9 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
.defaultValue(50)
.withDescription(
"Maximum number of HTTP connections in the S3
client connection pool. "
- + "Applies to both the sync client (Apache
HTTP) and the async client (Netty). "
+ + "Applies to the sync (Apache) and async
(Netty) clients. "
+ + "When s3.crt.enabled is true, the CRT
clients use "
+ + "'s3.crt.max-concurrency' instead of
this option. "
+ "Must be at least as large as
's3.bulk-copy.max-concurrent'.");
public static final ConfigOption<Integer> BULK_COPY_MAX_CONCURRENT =
@@ -161,6 +164,15 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
.defaultValue(16)
.withDescription("Maximum number of concurrent copy
operations");
+ public static final ConfigOption<Integer> BULK_COPY_DOWNLOAD_BUFFER_SIZE =
+ ConfigOptions.key("s3.bulk-copy.download-buffer-size")
+ .intType()
+ .defaultValue(256 * 1024) // 256KB default
+ .withDescription(
+ "Buffer size in bytes used when writing files
downloaded via bulk copy "
+ + "to the local filesystem. Bounds the
per-thread temporary "
+ + "direct buffers the JDK caches for
channel writes");
+
public static final ConfigOption<Boolean> USE_ASYNC_OPERATIONS =
ConfigOptions.key("s3.async.enabled")
.booleanType()
@@ -321,6 +333,69 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
+ "When not set, the default chain is
used: delegation tokens -> "
+ "static credentials (if configured) ->
DefaultCredentialsProvider.");
+ public static final ConfigOption<Boolean> CRT_ENABLED =
+ ConfigOptions.key("s3.crt.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Enable AWS Common Runtime (CRT) HTTP transport. "
+ + "When true, uses AwsCrtHttpClient for
sync S3 operations and "
+ + "S3AsyncClient.crtBuilder() for
async/transfer operations, "
+ + "providing higher throughput for large
S3 transfers. "
+ + "Requires the aws-crt JAR in the plugin
directory. "
+ + "The pure-Java aws-crt-client classes
are bundled in the fat JAR; "
+ + "only aws-crt is external because its
JNI classes cannot be shaded.");
+
+ public static final ConfigOption<Double> CRT_TARGET_THROUGHPUT_GBPS =
+ ConfigOptions.key("s3.crt.target-throughput-gbps")
+ .doubleType()
+ .noDefaultValue()
+ .withDescription(
+ "Soft target throughput in Gbps for the CRT-based
S3 async client. "
+ + "Only used when s3.crt.enabled is true. "
+ + "This is a hint to the CRT runtime, not
a hard cap: actual throughput "
+ + "may exceed this value, and the runtime
uses it to size its internal "
+ + "worker pool and tune parallelism, so
there is no separate Flink-level "
+ + "knob for concurrent CRT transfer
threads. "
+ + "When unset, the AWS CRT runtime applies
its own internal default. "
+ + "Set this only when you want to override
that — e.g. pick a value "
+ + "matching the network bandwidth
available to a single TaskManager.");
+
+ public static final ConfigOption<MemorySize> CRT_MAX_NATIVE_MEMORY_LIMIT =
+ ConfigOptions.key("s3.crt.max-native-memory-limit")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ "Maximum native memory the CRT-based S3 async
client may use. "
+ + "Only used when s3.crt.enabled is true. "
+ + "When unset, the AWS CRT runtime applies
its own internal limit.");
+
+ public static final ConfigOption<MemorySize> CRT_READ_BUFFER_SIZE =
+ ConfigOptions.key("s3.crt.read-buffer-size")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ "Read buffer size for the CRT HTTP transport
(applied to both the sync "
+ + "AwsCrtHttpClient and the async CRT
client's initial read buffer). "
+ + "Only used when s3.crt.enabled is true.
This is decoupled from "
+ + "'s3.read.buffer.size' so that lowering
the streaming read buffer "
+ + "does not shrink CRT's native transfer
buffers and regress throughput. "
+ + "When unset, the AWS CRT runtime applies
its own (larger) default.");
+
+ public static final ConfigOption<Integer> CRT_MAX_CONCURRENCY =
+ ConfigOptions.key("s3.crt.max-concurrency")
+ .intType()
+ .defaultValue(256)
+ .withDescription(
+ "Maximum number of concurrent requests the CRT
HTTP transport may have "
+ + "in flight (applied to both the sync
AwsCrtHttpClient and the "
+ + "async CRT client). Only used when
s3.crt.enabled is true. This "
+ + "is decoupled from 's3.connection.max'
because the CRT client "
+ + "fans a single logical transfer out into
many concurrent "
+ + "part-sized requests; reusing the
smaller sync connection-pool "
+ + "size here causes 'failed to acquire a
connection' timeouts "
+ + "under parallel checkpoint
upload/restore. Defaults to 256.");
+
@Nullable private Configuration flinkConfig;
@Nullable private BucketConfigProvider bucketConfigProvider;
@@ -462,7 +537,63 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
MAX_CONNECTIONS.key(),
maxConnections);
- S3ClientProvider clientProvider =
+ final boolean crtEnabled = config.get(CRT_ENABLED);
+
+ // CRT-only options. Validated only when CRT is enabled (mirroring the
bulk-copy gating
+ // below); when CRT is disabled these options are ignored. Unset
optional values fall back
+ // to the CRT runtime's own defaults.
+ final Double crtTargetThroughputGbps =
+ config.getOptional(CRT_TARGET_THROUGHPUT_GBPS).orElse(null);
+ final MemorySize crtMaxNativeMemoryLimit =
+ config.getOptional(CRT_MAX_NATIVE_MEMORY_LIMIT).orElse(null);
+ final MemorySize crtReadBufferSize =
config.getOptional(CRT_READ_BUFFER_SIZE).orElse(null);
+ final int crtMaxConcurrency = config.get(CRT_MAX_CONCURRENCY);
+ if (crtEnabled) {
+ if (crtTargetThroughputGbps != null) {
+ Preconditions.checkArgument(
+ crtTargetThroughputGbps > 0,
+ "'%s' must be positive, but was %s",
+ CRT_TARGET_THROUGHPUT_GBPS.key(),
+ crtTargetThroughputGbps);
+ }
+ if (crtMaxNativeMemoryLimit != null) {
+ Preconditions.checkArgument(
+ crtMaxNativeMemoryLimit.getBytes() > 0,
+ "'%s' must be positive, but was %s",
+ CRT_MAX_NATIVE_MEMORY_LIMIT.key(),
+ crtMaxNativeMemoryLimit);
+ }
+ if (crtReadBufferSize != null) {
+ Preconditions.checkArgument(
+ crtReadBufferSize.getBytes() > 0,
+ "'%s' must be positive, but was %s",
+ CRT_READ_BUFFER_SIZE.key(),
+ crtReadBufferSize);
+ }
+ Preconditions.checkArgument(
+ crtMaxConcurrency > 0,
+ "'%s' must be a positive integer, but was %s",
+ CRT_MAX_CONCURRENCY.key(),
+ crtMaxConcurrency);
+ }
+
+ final boolean bulkCopyEnabled = config.get(BULK_COPY_ENABLED);
+ final int bulkCopyMaxConcurrent = config.get(BULK_COPY_MAX_CONCURRENT);
+ final int bulkCopyDownloadBufferSize =
config.get(BULK_COPY_DOWNLOAD_BUFFER_SIZE);
+ if (bulkCopyEnabled) {
+ Preconditions.checkArgument(
+ bulkCopyMaxConcurrent > 0,
+ "'%s' must be a positive integer, but was %s",
+ BULK_COPY_MAX_CONCURRENT.key(),
+ bulkCopyMaxConcurrent);
+ Preconditions.checkArgument(
+ bulkCopyDownloadBufferSize > 0,
+ "'%s' must be a positive integer, but was %s",
+ BULK_COPY_DOWNLOAD_BUFFER_SIZE.key(),
+ bulkCopyDownloadBufferSize);
+ }
+
+ S3ClientProvider.Builder clientProviderBuilder =
S3ClientProvider.builder()
.accessKey(accessKey)
.secretKey(secretKey)
@@ -486,21 +617,30 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
.retryMaxBackoff(config.get(RETRY_MAX_BACKOFF))
.credentialsProviderClasses(credentialsProviderClasses)
.encryptionConfig(encryptionConfig)
- .build();
+ .useCrt(crtEnabled);
+
+ if (crtEnabled) {
+ clientProviderBuilder
+ .crtTargetThroughputGbps(crtTargetThroughputGbps)
+ .crtReadBufferSizeInBytes(
+ crtReadBufferSize == null ? null :
crtReadBufferSize.getBytes())
+ .crtMaxConcurrency(crtMaxConcurrency)
+ .crtMaxNativeMemoryLimitInBytes(
+ crtMaxNativeMemoryLimit == null
+ ? null
+ : crtMaxNativeMemoryLimit.getBytes())
+ .crtMinPartSizeInBytes(config.get(PART_UPLOAD_MIN_SIZE));
+ }
+ S3ClientProvider clientProvider = clientProviderBuilder.build();
NativeS3BulkCopyHelper bulkCopyHelper = null;
- if (config.get(BULK_COPY_ENABLED)) {
- final int bulkCopyMaxConcurrent =
config.get(BULK_COPY_MAX_CONCURRENT);
- Preconditions.checkArgument(
- bulkCopyMaxConcurrent > 0,
- "'%s' must be a positive integer, but was %s",
- BULK_COPY_MAX_CONCURRENT.key(),
- bulkCopyMaxConcurrent);
+ if (bulkCopyEnabled) {
bulkCopyHelper =
new NativeS3BulkCopyHelper(
- clientProvider.getTransferManager(),
+ clientProvider.getAsyncClient(),
bulkCopyMaxConcurrent,
- maxConnections);
+ maxConnections,
+ bulkCopyDownloadBufferSize);
}
return new NativeS3FileSystem(
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3OutputStream.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3OutputStream.java
index d9a123d1b53..94a26b87f84 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3OutputStream.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3OutputStream.java
@@ -79,9 +79,7 @@ class NativeS3OutputStream extends FSDataOutputStream {
Preconditions.checkNotNull(encryptionConfig, "encryptionConfig
must not be null");
File tmpDir = new File(localTmpDir);
- if (!tmpDir.exists()) {
- tmpDir.mkdirs();
- }
+ Files.createDirectories(tmpDir.toPath());
this.tmpFile = new File(tmpDir, "s3-upload-" + UUID.randomUUID());
this.bufferedStream = new BufferedOutputStream(new
FileOutputStream(tmpFile), BUFFER_SIZE);
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
index e3c7a1f2380..a12ef1da405 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
@@ -34,6 +34,7 @@ import
software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.http.crt.AwsCrtHttpClient;
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain;
@@ -44,6 +45,9 @@ import
software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.S3ClientBuilder;
import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder;
+import software.amazon.awssdk.services.s3.crt.S3CrtHttpConfiguration;
+import software.amazon.awssdk.services.s3.crt.S3CrtRetryConfiguration;
import software.amazon.awssdk.services.sts.StsClient;
import
software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
@@ -76,6 +80,7 @@ class S3ClientProvider implements AutoCloseableAsync {
private static final Logger LOG =
LoggerFactory.getLogger(S3ClientProvider.class);
private final S3Client s3Client;
+ private final S3AsyncClient asyncClient;
private final S3TransferManager transferManager;
private final S3EncryptionConfig encryptionConfig;
private final AwsCredentialsProvider credentialsProvider;
@@ -98,10 +103,16 @@ class S3ClientProvider implements AutoCloseableAsync {
@Nullable private final String assumeRoleExternalId;
@Nullable private final String assumeRoleSessionName;
private final int assumeRoleSessionDurationSeconds;
+ private final boolean useCrt;
+ @Nullable private final Double crtTargetThroughputGbps;
+ @Nullable private final Long crtReadBufferSizeInBytes;
+ private final int crtMaxConcurrency;
+ @Nullable private final Long crtMaxNativeMemoryLimitInBytes;
private final AtomicBoolean closed = new AtomicBoolean(false);
private S3ClientProvider(
S3Client s3Client,
+ S3AsyncClient asyncClient,
S3TransferManager transferManager,
S3EncryptionConfig encryptionConfig,
AwsCredentialsProvider credentialsProvider,
@@ -123,8 +134,14 @@ class S3ClientProvider implements AutoCloseableAsync {
@Nullable String assumeRoleArn,
@Nullable String assumeRoleExternalId,
@Nullable String assumeRoleSessionName,
- int assumeRoleSessionDurationSeconds) {
+ int assumeRoleSessionDurationSeconds,
+ boolean useCrt,
+ @Nullable Double crtTargetThroughputGbps,
+ @Nullable Long crtReadBufferSizeInBytes,
+ int crtMaxConcurrency,
+ @Nullable Long crtMaxNativeMemoryLimitInBytes) {
this.s3Client = Preconditions.checkNotNull(s3Client, "s3Client must
not be null");
+ this.asyncClient = Preconditions.checkNotNull(asyncClient,
"asyncClient must not be null");
this.transferManager =
Preconditions.checkNotNull(transferManager, "transferManager
must not be null");
this.encryptionConfig =
@@ -161,6 +178,11 @@ class S3ClientProvider implements AutoCloseableAsync {
this.assumeRoleExternalId = assumeRoleExternalId;
this.assumeRoleSessionName = assumeRoleSessionName;
this.assumeRoleSessionDurationSeconds =
assumeRoleSessionDurationSeconds;
+ this.useCrt = useCrt;
+ this.crtTargetThroughputGbps = crtTargetThroughputGbps;
+ this.crtReadBufferSizeInBytes = crtReadBufferSizeInBytes;
+ this.crtMaxConcurrency = crtMaxConcurrency;
+ this.crtMaxNativeMemoryLimitInBytes = crtMaxNativeMemoryLimitInBytes;
}
public S3Client getS3Client() {
@@ -283,6 +305,44 @@ class S3ClientProvider implements AutoCloseableAsync {
return assumeRoleSessionDurationSeconds;
}
+ @VisibleForTesting
+ boolean isUseCrt() {
+ return useCrt;
+ }
+
+ @VisibleForTesting
+ @Nullable
+ Double getCrtTargetThroughputGbps() {
+ return crtTargetThroughputGbps;
+ }
+
+ @VisibleForTesting
+ @Nullable
+ Long getCrtReadBufferSizeInBytes() {
+ return crtReadBufferSizeInBytes;
+ }
+
+ @VisibleForTesting
+ int getCrtMaxConcurrency() {
+ return crtMaxConcurrency;
+ }
+
+ @VisibleForTesting
+ @Nullable
+ Long getCrtMaxNativeMemoryLimitInBytes() {
+ return crtMaxNativeMemoryLimitInBytes;
+ }
+
+ /**
+ * Exposed for tests to verify which async-client implementation (CRT vs
Netty) was actually
+ * constructed. Not intended for production use; callers should go through
{@link
+ * #getTransferManager()}.
+ */
+ @VisibleForTesting
+ S3AsyncClient getAsyncClient() {
+ return asyncClient;
+ }
+
@Override
public CompletableFuture<Void> closeAsync() {
if (!closed.compareAndSet(false, true)) {
@@ -295,6 +355,11 @@ class S3ClientProvider implements AutoCloseableAsync {
} catch (Exception e) {
LOG.warn("Error closing S3 TransferManager",
e);
}
+ try {
+ asyncClient.close();
+ } catch (Exception e) {
+ LOG.warn("Error closing S3 async client", e);
+ }
try {
s3Client.close();
} catch (Exception e) {
@@ -315,10 +380,13 @@ class S3ClientProvider implements AutoCloseableAsync {
}
}
})
- .orTimeout(clientCloseTimeout.toSeconds(), TimeUnit.SECONDS)
+ .orTimeout(clientCloseTimeout.toMillis(),
TimeUnit.MILLISECONDS)
.exceptionally(
ex -> {
- LOG.error("S3 client close timed out after {}",
clientCloseTimeout, ex);
+ LOG.error(
+ "S3 client close did not complete cleanly
within {}",
+ clientCloseTimeout,
+ ex);
return null;
});
}
@@ -338,26 +406,37 @@ class S3ClientProvider implements AutoCloseableAsync {
private String secretKey;
private String region;
private String endpoint;
- private boolean pathStyleAccess = false;
- private boolean chunkedEncoding = true;
- private boolean checksumValidation = true;
- private int maxConnections = 50;
- private Duration connectionTimeout = Duration.ofSeconds(60);
- private Duration socketTimeout = Duration.ofSeconds(60);
- private Duration connectionMaxIdleTime = Duration.ofSeconds(60);
- private int maxRetries = 3;
+ // All defaults are sourced from NativeS3FileSystemFactory
ConfigOption.defaultValue() so
+ // that NativeS3FileSystemFactory remains the single source of truth —
if a default changes
+ // there, this Builder automatically picks it up without needing a
parallel edit.
+ private boolean pathStyleAccess =
+ NativeS3FileSystemFactory.PATH_STYLE_ACCESS.defaultValue();
+ private boolean chunkedEncoding =
+
NativeS3FileSystemFactory.CHUNKED_ENCODING_ENABLED.defaultValue();
+ private boolean checksumValidation =
+
NativeS3FileSystemFactory.CHECKSUM_VALIDATION_ENABLED.defaultValue();
+ private int maxConnections =
NativeS3FileSystemFactory.MAX_CONNECTIONS.defaultValue();
+ private Duration connectionTimeout =
+ NativeS3FileSystemFactory.CONNECTION_TIMEOUT.defaultValue();
+ private Duration socketTimeout =
NativeS3FileSystemFactory.SOCKET_TIMEOUT.defaultValue();
+ private Duration connectionMaxIdleTime =
+
NativeS3FileSystemFactory.CONNECTION_MAX_IDLE_TIME.defaultValue();
+ private int maxRetries =
NativeS3FileSystemFactory.MAX_RETRIES.defaultValue();
private Duration retryBaseDelay =
NativeS3FileSystemFactory.RETRY_BASE_DELAY.defaultValue();
private Duration retryThrottleBaseDelay =
NativeS3FileSystemFactory.RETRY_THROTTLE_BASE_DELAY.defaultValue();
private Duration retryMaxBackoff =
NativeS3FileSystemFactory.RETRY_MAX_BACKOFF.defaultValue();
- private Duration clientCloseTimeout = Duration.ofSeconds(30);
+ private Duration clientCloseTimeout =
+ NativeS3FileSystemFactory.CLIENT_CLOSE_TIMEOUT.defaultValue();
// AssumeRole configuration
private String assumeRoleArn;
private String assumeRoleExternalId;
- private String assumeRoleSessionName = "flink-s3-session";
- private int assumeRoleSessionDurationSeconds = 3600;
+ private String assumeRoleSessionName =
+
NativeS3FileSystemFactory.ASSUME_ROLE_SESSION_NAME.defaultValue();
+ private int assumeRoleSessionDurationSeconds =
+
NativeS3FileSystemFactory.ASSUME_ROLE_SESSION_DURATION_SECONDS.defaultValue();
// Encryption configuration
private S3EncryptionConfig encryptionConfig =
S3EncryptionConfig.none();
@@ -365,6 +444,16 @@ class S3ClientProvider implements AutoCloseableAsync {
// Custom credentials provider class names (comma-separated)
@Nullable private String credentialsProviderClasses;
+ // CRT configuration
+ private boolean useCrt =
NativeS3FileSystemFactory.CRT_ENABLED.defaultValue();
+ @Nullable private Double crtTargetThroughputGbps = null;
+ @Nullable private Long crtReadBufferSizeInBytes = null;
+ private int crtMaxConcurrency =
+ NativeS3FileSystemFactory.CRT_MAX_CONCURRENCY.defaultValue();
+ @Nullable private Long crtMaxNativeMemoryLimitInBytes = null;
+ private long crtMinPartSizeInBytes =
+ NativeS3FileSystemFactory.PART_UPLOAD_MIN_SIZE.defaultValue();
+
public Builder accessKey(@Nullable String accessKey) {
this.accessKey = accessKey;
return this;
@@ -498,6 +587,37 @@ class S3ClientProvider implements AutoCloseableAsync {
return this;
}
+ public Builder useCrt(boolean useCrt) {
+ this.useCrt = useCrt;
+ return this;
+ }
+
+ public Builder crtTargetThroughputGbps(@Nullable Double
crtTargetThroughputGbps) {
+ this.crtTargetThroughputGbps = crtTargetThroughputGbps;
+ return this;
+ }
+
+ public Builder crtReadBufferSizeInBytes(@Nullable Long
crtReadBufferSizeInBytes) {
+ this.crtReadBufferSizeInBytes = crtReadBufferSizeInBytes;
+ return this;
+ }
+
+ public Builder crtMaxConcurrency(int crtMaxConcurrency) {
+ this.crtMaxConcurrency = crtMaxConcurrency;
+ return this;
+ }
+
+ public Builder crtMaxNativeMemoryLimitInBytes(
+ @Nullable Long crtMaxNativeMemoryLimitInBytes) {
+ this.crtMaxNativeMemoryLimitInBytes =
crtMaxNativeMemoryLimitInBytes;
+ return this;
+ }
+
+ public Builder crtMinPartSizeInBytes(long crtMinPartSizeInBytes) {
+ this.crtMinPartSizeInBytes = crtMinPartSizeInBytes;
+ return this;
+ }
+
S3ClientProvider build() {
if (endpoint == null) {
endpoint = System.getProperty("s3.endpoint");
@@ -554,46 +674,27 @@ class S3ClientProvider implements AutoCloseableAsync {
.build())
.build();
- ApacheHttpClient.Builder httpClientBuilder =
- ApacheHttpClient.builder()
- .maxConnections(maxConnections)
- .connectionTimeout(connectionTimeout)
- .socketTimeout(socketTimeout)
- .tcpKeepAlive(true)
- .connectionMaxIdleTime(connectionMaxIdleTime);
-
- S3ClientBuilder clientBuilder =
- S3Client.builder()
- .credentialsProvider(credentialsProvider)
- .region(awsRegion)
- .serviceConfiguration(s3Config)
- .httpClientBuilder(httpClientBuilder)
- .overrideConfiguration(overrideConfig);
- if (endpointUri != null) {
- clientBuilder.endpointOverride(endpointUri);
+ if (useCrt) {
+ LOG.info(
+ "AWS CRT transport enabled (s3.crt.enabled=true) with
target throughput {}",
+ crtTargetThroughputGbps != null
+ ? crtTargetThroughputGbps + " Gbps"
+ : "(CRT runtime default)");
}
- S3Client s3Client = clientBuilder.build();
- S3AsyncClientBuilder asyncClientBuilder =
- S3AsyncClient.builder()
- .credentialsProvider(credentialsProvider)
- .region(awsRegion)
- .serviceConfiguration(s3Config)
- .httpClientBuilder(
- NettyNioAsyncHttpClient.builder()
- .maxConcurrency(maxConnections)
-
.connectionTimeout(connectionTimeout)
- .readTimeout(socketTimeout)
-
.connectionAcquisitionTimeout(connectionTimeout))
- .overrideConfiguration(overrideConfig);
- if (endpointUri != null) {
- asyncClientBuilder.endpointOverride(endpointUri);
- }
+ S3Client s3Client =
+ buildSyncClient(
+ credentialsProvider, awsRegion, s3Config,
overrideConfig, endpointUri);
+ S3AsyncClient asyncClient =
+ buildAsyncClient(
+ credentialsProvider, awsRegion, s3Config,
overrideConfig, endpointUri);
+
S3TransferManager transferManager =
-
S3TransferManager.builder().s3Client(asyncClientBuilder.build()).build();
+ S3TransferManager.builder().s3Client(asyncClient).build();
return new S3ClientProvider(
s3Client,
+ asyncClient,
transferManager,
encryptionConfig,
credentialsProvider,
@@ -615,7 +716,161 @@ class S3ClientProvider implements AutoCloseableAsync {
assumeRoleArn,
assumeRoleExternalId,
assumeRoleSessionName,
- assumeRoleSessionDurationSeconds);
+ assumeRoleSessionDurationSeconds,
+ useCrt,
+ crtTargetThroughputGbps,
+ crtReadBufferSizeInBytes,
+ crtMaxConcurrency,
+ crtMaxNativeMemoryLimitInBytes);
+ }
+
+ /**
+ * Builds the synchronous {@link S3Client}, choosing between the
Apache HTTP transport and
+ * the AWS CRT transport based on {@link #useCrt}.
+ */
+ private S3Client buildSyncClient(
+ AwsCredentialsProvider credentialsProvider,
+ Region awsRegion,
+ S3Configuration s3Config,
+ ClientOverrideConfiguration overrideConfig,
+ @Nullable URI endpointUri) {
+ S3ClientBuilder clientBuilder =
+ S3Client.builder()
+ .credentialsProvider(credentialsProvider)
+ .region(awsRegion)
+ .serviceConfiguration(s3Config)
+ .overrideConfiguration(overrideConfig);
+
+ try {
+ if (useCrt) {
+ // Note: AwsCrtHttpClient.Builder does not expose a
`readTimeout(Duration)`
+ // equivalent of the Apache client's socket timeout. The
CRT runtime relies
+ // on `ConnectionHealthConfiguration` for stalled-read
detection instead, so
+ // the `s3.socket.timeout` setting is silently ignored in
CRT mode.
+ AwsCrtHttpClient.Builder crtHttpBuilder =
+ AwsCrtHttpClient.builder()
+ .maxConcurrency(crtMaxConcurrency)
+ .connectionTimeout(connectionTimeout)
+
.connectionMaxIdleTime(connectionMaxIdleTime);
+ if (crtReadBufferSizeInBytes != null) {
+
crtHttpBuilder.readBufferSizeInBytes(crtReadBufferSizeInBytes);
+ }
+ clientBuilder.httpClientBuilder(crtHttpBuilder);
+ } else {
+ clientBuilder.httpClientBuilder(
+ ApacheHttpClient.builder()
+ .maxConnections(maxConnections)
+ .connectionTimeout(connectionTimeout)
+ .socketTimeout(socketTimeout)
+ .tcpKeepAlive(true)
+
.connectionMaxIdleTime(connectionMaxIdleTime));
+ }
+ if (endpointUri != null) {
+ clientBuilder.endpointOverride(endpointUri);
+ }
+ return clientBuilder.build();
+ } catch (LinkageError e) {
+ if (useCrt) {
+ throw new IllegalStateException(crtMissingJarsMessage(),
e);
+ }
+ throw e;
+ } catch (IllegalStateException e) {
+ if (useCrt && isCrtClasspathFailure(e)) {
+ throw new IllegalStateException(crtMissingJarsMessage(),
e);
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Builds the asynchronous {@link S3AsyncClient}, choosing between the
Netty transport and
+ * the AWS CRT transport based on {@link #useCrt}.
+ *
+ * <p>Note: when CRT is enabled the {@code
s3.chunked-encoding.enabled} option is silently
+ * ignored. The CRT runtime manages wire encoding internally and {@link
+ * software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder} exposes
no equivalent setter.
+ */
+ private S3AsyncClient buildAsyncClient(
+ AwsCredentialsProvider credentialsProvider,
+ Region awsRegion,
+ S3Configuration s3Config,
+ ClientOverrideConfiguration overrideConfig,
+ @Nullable URI endpointUri) {
+ if (useCrt) {
+ try {
+ S3CrtAsyncClientBuilder crtAsyncBuilder =
+ S3AsyncClient.crtBuilder()
+ .credentialsProvider(credentialsProvider)
+ .region(awsRegion)
+ .forcePathStyle(pathStyleAccess)
+
.checksumValidationEnabled(checksumValidation)
+ .httpConfiguration(
+ S3CrtHttpConfiguration.builder()
+
.connectionTimeout(connectionTimeout)
+ .build())
+ .retryConfiguration(
+ S3CrtRetryConfiguration.builder()
+ .numRetries(maxRetries)
+ .build())
+ .maxConcurrency(crtMaxConcurrency)
+
.minimumPartSizeInBytes(crtMinPartSizeInBytes);
+ if (crtReadBufferSizeInBytes != null) {
+
crtAsyncBuilder.initialReadBufferSizeInBytes(crtReadBufferSizeInBytes);
+ }
+ // Only override the CRT runtime's own default when the
user has explicitly
+ // configured s3.crt.target-throughput-gbps; otherwise let
the SDK pick.
+ if (crtTargetThroughputGbps != null) {
+
crtAsyncBuilder.targetThroughputInGbps(crtTargetThroughputGbps);
+ }
+ if (crtMaxNativeMemoryLimitInBytes != null) {
+
crtAsyncBuilder.maxNativeMemoryLimitInBytes(crtMaxNativeMemoryLimitInBytes);
+ }
+ if (endpointUri != null) {
+ crtAsyncBuilder.endpointOverride(endpointUri);
+ }
+ return crtAsyncBuilder.build();
+ } catch (LinkageError e) {
+ throw new IllegalStateException(crtMissingJarsMessage(),
e);
+ } catch (IllegalStateException e) {
+ if (isCrtClasspathFailure(e)) {
+ throw new
IllegalStateException(crtMissingJarsMessage(), e);
+ }
+ throw e;
+ }
+ }
+
+ S3AsyncClientBuilder asyncBuilder =
+ S3AsyncClient.builder()
+ .credentialsProvider(credentialsProvider)
+ .region(awsRegion)
+ .serviceConfiguration(s3Config)
+ .httpClientBuilder(
+ NettyNioAsyncHttpClient.builder()
+ .maxConcurrency(maxConnections)
+
.connectionTimeout(connectionTimeout)
+ .readTimeout(socketTimeout)
+
.connectionAcquisitionTimeout(connectionTimeout))
+ .overrideConfiguration(overrideConfig);
+ if (endpointUri != null) {
+ asyncBuilder.endpointOverride(endpointUri);
+ }
+ return asyncBuilder.build();
+ }
+
+ private static boolean isCrtClasspathFailure(IllegalStateException e) {
+ String message = e.getMessage();
+ return message != null
+ && (message.contains("AWS Common Runtime")
+ || message.contains("software.amazon.awssdk.crt"));
+ }
+
+ @VisibleForTesting
+ static String crtMissingJarsMessage() {
+ return "CRT transport requested (s3.crt.enabled=true) but the
aws-crt JAR "
+ + "is not on the classpath. Place it in the Flink plugin
directory "
+ + "(e.g. $FLINK_HOME/plugins/s3-fs-native/) alongside
flink-s3-fs-native.jar. "
+ + "Run tools/download-crt-jars.sh to download the matching
version. "
+ + "See the module README for setup details.";
}
private AwsCredentialsProvider buildBaseCredentialsProvider() {
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java
index 14ff1eeab12..1a8fd361743 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java
@@ -20,13 +20,15 @@ package org.apache.flink.fs.s3native.writer;
import org.apache.flink.annotation.Internal;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.fs.s3native.NativeS3FileIoUtils;
import org.apache.flink.fs.s3native.S3EncryptionConfig;
import org.apache.flink.fs.s3native.S3ExceptionUtils;
+import org.apache.flink.util.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.ResponseInputStream;
import software.amazon.awssdk.core.sync.RequestBody;
-import software.amazon.awssdk.core.sync.ResponseTransformer;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
@@ -55,6 +57,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.List;
+import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
/**
@@ -97,6 +100,8 @@ public class NativeS3ObjectOperations {
private static final Logger LOG =
LoggerFactory.getLogger(NativeS3ObjectOperations.class);
+ private static final int DOWNLOAD_BUFFER_SIZE = 256 * 1024;
+
private final S3Client s3Client;
private final S3TransferManager transferManager;
private final String bucketName;
@@ -259,8 +264,20 @@ public class NativeS3ObjectOperations {
.build();
FileUpload fileUpload = transferManager.uploadFile(uploadRequest);
- CompletedFileUpload completedUpload =
fileUpload.completionFuture().join();
+ CompletedFileUpload completedUpload;
+ try {
+ completedUpload = fileUpload.completionFuture().get();
+ } catch (InterruptedException e) {
+ fileUpload.completionFuture().cancel(true);
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while uploading object for
key: " + key, e);
+ } catch (ExecutionException e) {
+ throw new IOException(
+ "Failed to async upload object for key: " + key,
e.getCause());
+ }
return new PutObjectResult(completedUpload.response().eTag());
+ } catch (IOException e) {
+ throw e;
} catch (Exception e) {
throw new IOException("Failed to async upload object for key: " +
key, e);
}
@@ -380,16 +397,45 @@ public class NativeS3ObjectOperations {
}
public long getObject(String key, File targetLocation) throws IOException {
+ java.nio.file.Path target = targetLocation.toPath().toAbsolutePath();
+ java.nio.file.Path parent = target.getParent();
+ if (parent != null) {
+ Files.createDirectories(parent);
+ }
+ java.nio.file.Path tempTarget =
+ NativeS3FileIoUtils.createTemporaryDownloadFile(parent,
target);
+ ResponseInputStream<GetObjectResponse> responseStream = null;
+ boolean success = false;
try {
GetObjectRequest request =
GetObjectRequest.builder().bucket(bucketName).key(key).build();
- ResponseTransformer<GetObjectResponse, GetObjectResponse>
responseTransformer =
- ResponseTransformer.toFile(targetLocation.toPath());
- s3Client.getObject(request, responseTransformer);
- return Files.size(targetLocation.toPath());
+ responseStream = s3Client.getObject(request);
+ NativeS3FileIoUtils.copyStream(responseStream, tempTarget,
DOWNLOAD_BUFFER_SIZE);
+ NativeS3FileIoUtils.moveFile(tempTarget, target);
+ success = true;
+ return Files.size(target);
} catch (S3Exception e) {
throw new IOException("Failed to get object for key: " + key, e);
+ } finally {
+ if (success) {
+ IOUtils.closeQuietly(responseStream);
+ } else {
+ abortAndClose(responseStream);
+ IOUtils.deleteFileQuietly(tempTarget);
+ }
+ }
+ }
+
+ private static void abortAndClose(ResponseInputStream<GetObjectResponse>
stream) {
+ if (stream == null) {
+ return;
+ }
+ try {
+ stream.abort();
+ } catch (RuntimeException e) {
+ LOG.debug("Error aborting S3 response stream", e);
}
+ IOUtils.closeQuietly(stream);
}
public ObjectMetadata getObjectMetadata(String key) throws IOException {
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
index 7eabcb9db2b..0a0376fcfac 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
@@ -21,6 +21,7 @@ package org.apache.flink.fs.s3native.writer;
import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.fs.s3native.writer.NativeS3Recoverable.PartETag;
+import org.apache.flink.util.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -128,9 +129,7 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
private void createNewTempFile() throws IOException {
File tmpDir = new File(localTmpDir);
- if (!tmpDir.exists()) {
- tmpDir.mkdirs();
- }
+ Files.createDirectories(tmpDir.toPath());
currentTempFile = new File(tmpDir, "s3-part-" + UUID.randomUUID());
currentFileStream = new FileOutputStream(currentTempFile);
@@ -198,11 +197,14 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
private void uploadCurrentPart() throws IOException {
currentOutputStream.close();
- int partNumber = nextPartNumber++;
+ // Do not delete the temp file if uploadPart fails: propagate the
original exception
+ // unmasked and let close() perform cleanup. nextPartNumber is only
advanced on success so a
+ // failed attempt does not leave a gap in the part sequence.
NativeS3ObjectOperations.UploadPartResult result =
s3AccessHelper.uploadPart(
- key, uploadId, partNumber, currentTempFile,
currentPartSize);
+ key, uploadId, nextPartNumber, currentTempFile,
currentPartSize);
+ nextPartNumber++;
completedParts.add(new PartETag(result.getPartNumber(),
result.getETag()));
numBytesInParts += currentPartSize;
@@ -217,7 +219,6 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
throw new IOException("Stream is already closed");
}
- closed = true;
currentOutputStream.close();
if (currentPartSize > 0) {
@@ -230,6 +231,7 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
new NativeS3Recoverable(
key, uploadId, new ArrayList<>(completedParts),
numBytesInParts);
+ closed = true;
return new NativeS3Committer(s3AccessHelper, recoverable);
} finally {
unlock();
@@ -270,11 +272,20 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
try {
if (!closed) {
closed = true;
+ IOException cleanupException = null;
if (currentOutputStream != null) {
- currentOutputStream.close();
+ try {
+ currentOutputStream.close();
+ } catch (IOException e) {
+ cleanupException = ExceptionUtils.firstOrSuppressed(e,
cleanupException);
+ }
}
if (currentTempFile != null && currentTempFile.exists()) {
- Files.delete(currentTempFile.toPath());
+ try {
+ Files.delete(currentTempFile.toPath());
+ } catch (IOException e) {
+ cleanupException = ExceptionUtils.firstOrSuppressed(e,
cleanupException);
+ }
}
try {
@@ -287,6 +298,9 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
uploadId,
e);
}
+ if (cleanupException != null) {
+ throw cleanupException;
+ }
}
} finally {
unlock();
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
b/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
index 427b561089b..a91513a5175 100644
--- a/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
+++ b/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
@@ -11,6 +11,7 @@ This project bundles the following dependencies under the
Apache Software Licens
- software.amazon.awssdk:arns:2.44.4
- software.amazon.awssdk:auth:2.44.4
- software.amazon.awssdk:aws-core:2.44.4
+- software.amazon.awssdk:aws-crt-client:2.44.4
- software.amazon.awssdk:aws-query-protocol:2.44.4
- software.amazon.awssdk:aws-xml-protocol:2.44.4
- software.amazon.awssdk:checksums:2.44.4
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelperTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelperTest.java
index 5b244850dca..d8325d9aba5 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelperTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelperTest.java
@@ -18,24 +18,44 @@
package org.apache.flink.fs.s3native;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.PathsCopyingFileSystem.CopyRequest;
+
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.core.ResponseInputStream;
import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Proxy;
+import java.nio.file.Path;
import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
+import static
org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.assertj.core.api.Assertions.assertThatNullPointerException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link NativeS3BulkCopyHelper}. */
class NativeS3BulkCopyHelperTest {
- private static final NativeS3BulkCopyHelper helper = new
NativeS3BulkCopyHelper(null, 1, 1);
+ private static final NativeS3BulkCopyHelper helper =
+ new NativeS3BulkCopyHelper(null, 1, 1, 256 * 1024);
// --- URI parsing tests ---
@@ -153,7 +173,309 @@ class NativeS3BulkCopyHelperTest {
@Test
void testEmptyRequestListIsNoOp() throws Exception {
- NativeS3BulkCopyHelper noOpHelper = new NativeS3BulkCopyHelper(null,
16, 50);
+ NativeS3BulkCopyHelper noOpHelper = new NativeS3BulkCopyHelper(null,
16, 50, 256 * 1024);
noOpHelper.copyFiles(Collections.emptyList(), null);
}
+
+ // --- download buffer size tests ---
+
+ @Test
+ void testDownloadBufferSizeIsExposed() {
+ NativeS3BulkCopyHelper h = new NativeS3BulkCopyHelper(null, 1, 1, 128
* 1024);
+ assertThat(h.getDownloadBufferSize()).isEqualTo(128 * 1024);
+ }
+
+ @Test
+ void testNonPositiveDownloadBufferSizeRejected() {
+ assertThatIllegalArgumentException()
+ .isThrownBy(() -> new NativeS3BulkCopyHelper(null, 1, 1, 0));
+ }
+
+ @Test
+ void testCopyFilesCancellationClosesActiveResponseStream(@TempDir Path
tempDir)
+ throws Exception {
+ BlockingInputStream blockingStream = new BlockingInputStream();
+ ResponseInputStream<GetObjectResponse> responseStream =
+ new ResponseInputStream<>(GetObjectResponse.builder().build(),
blockingStream);
+ NativeS3BulkCopyHelper h =
+ new NativeS3BulkCopyHelper(
+
asyncClientReturning(CompletableFuture.completedFuture(responseStream)),
+ 1,
+ 1,
+ 1024);
+ CloseableRegistry registry = new CloseableRegistry();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CompletableFuture<Throwable> copyFailure = new CompletableFuture<>();
+
+ try {
+ executor.submit(
+ () -> {
+ try {
+ h.copyFiles(
+ Collections.singletonList(
+ CopyRequest.of(
+ new
org.apache.flink.core.fs.Path(
+ "s3://bucket/key"),
+ new
org.apache.flink.core.fs.Path(
+
tempDir.resolve("out").toUri()),
+ 1L)),
+ registry);
+ copyFailure.complete(null);
+ } catch (Throwable t) {
+ copyFailure.complete(t);
+ }
+ });
+
+ assertThat(blockingStream.awaitReadStarted()).isTrue();
+ registry.close();
+
+ assertThat(copyFailure.get(10,
TimeUnit.SECONDS)).isInstanceOf(IOException.class);
+ assertThat(blockingStream.awaitClosed()).isTrue();
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testCopyFilesClosesResponseStreamCompletedAfterCancellation(@TempDir
Path tempDir)
+ throws Exception {
+ BlockingInputStream blockingStream = new BlockingInputStream();
+ ResponseInputStream<GetObjectResponse> responseStream =
+ new ResponseInputStream<>(GetObjectResponse.builder().build(),
blockingStream);
+ NonCancellingCompletableFuture<ResponseInputStream<GetObjectResponse>>
responseFuture =
+ new NonCancellingCompletableFuture<>();
+ CountDownLatch getObjectCalled = new CountDownLatch(1);
+ NativeS3BulkCopyHelper h =
+ new NativeS3BulkCopyHelper(
+ asyncClientReturning(responseFuture, getObjectCalled),
1, 1, 1024);
+ CloseableRegistry registry = new CloseableRegistry();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CompletableFuture<Throwable> copyFailure = new CompletableFuture<>();
+
+ try {
+ executor.submit(
+ () -> {
+ try {
+ h.copyFiles(
+ Collections.singletonList(
+ CopyRequest.of(
+ new
org.apache.flink.core.fs.Path(
+ "s3://bucket/key"),
+ new
org.apache.flink.core.fs.Path(
+
tempDir.resolve("out").toUri()),
+ 1L)),
+ registry);
+ copyFailure.complete(null);
+ } catch (Throwable t) {
+ copyFailure.complete(t);
+ }
+ });
+
+ assertThat(getObjectCalled.await(10, TimeUnit.SECONDS)).isTrue();
+ registry.close();
+ responseFuture.complete(responseStream);
+
+ assertThat(copyFailure.get(10,
TimeUnit.SECONDS)).isInstanceOf(IOException.class);
+ assertThat(blockingStream.awaitClosed()).isTrue();
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testCopyFilesDeletesTemporaryFileWhenRequestFails(@TempDir Path
tempDir) {
+ NativeS3BulkCopyHelper h =
+ new NativeS3BulkCopyHelper(
+ asyncClientReturning(
+ CompletableFuture.failedFuture(new
IOException("boom"))),
+ 1,
+ 1,
+ 1024);
+
+ assertThatThrownBy(
+ () ->
+ h.copyFiles(
+ Collections.singletonList(
+ CopyRequest.of(
+ new
org.apache.flink.core.fs.Path(
+
"s3://bucket/key"),
+ new
org.apache.flink.core.fs.Path(
+
tempDir.resolve("out").toUri()),
+ 1L)),
+ new CloseableRegistry()))
+ .isInstanceOf(IOException.class);
+
+ assertThat(tempDir).isEmptyDirectory();
+ }
+
+ @Test
+ void testCopyFilesSupportsShortDestinationFileNames(@TempDir Path tempDir)
throws Exception {
+ ResponseInputStream<GetObjectResponse> responseStream =
+ new ResponseInputStream<>(
+ GetObjectResponse.builder().build(),
+ new ByteArrayInputStream("data".getBytes()));
+ NativeS3BulkCopyHelper h =
+ new NativeS3BulkCopyHelper(
+
asyncClientReturning(CompletableFuture.completedFuture(responseStream)),
+ 1,
+ 1,
+ 1024);
+ Path destination = tempDir.resolve("x");
+
+ h.copyFiles(
+ Collections.singletonList(
+ CopyRequest.of(
+ new
org.apache.flink.core.fs.Path("s3://bucket/key"),
+ new
org.apache.flink.core.fs.Path(destination.toUri()),
+ 4L)),
+ new CloseableRegistry());
+
+ assertThat(destination).hasContent("data");
+ }
+
+ @Test
+ void testCopyFilesFailFastAbortsInFlightStreamsOnFirstFailure(@TempDir
Path tempDir)
+ throws Exception {
+ BlockingInputStream blockingStream = new BlockingInputStream();
+ ResponseInputStream<GetObjectResponse> blockingResponse =
+ new ResponseInputStream<>(GetObjectResponse.builder().build(),
blockingStream);
+ java.util.concurrent.ConcurrentLinkedQueue<
+
CompletableFuture<ResponseInputStream<GetObjectResponse>>>
+ responses = new java.util.concurrent.ConcurrentLinkedQueue<>();
+ responses.add(CompletableFuture.completedFuture(blockingResponse));
+ responses.add(CompletableFuture.failedFuture(new IOException("boom")));
+
+ S3AsyncClient client =
+ (S3AsyncClient)
+ Proxy.newProxyInstance(
+ S3AsyncClient.class.getClassLoader(),
+ new Class<?>[] {S3AsyncClient.class},
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "getObject":
+ return responses.poll();
+ case "close":
+ return null;
+ case "serviceName":
+ return "s3";
+ case "toString":
+ return "test-s3-async-client";
+ default:
+ throw new
UnsupportedOperationException(
+ method.toString());
+ }
+ });
+ NativeS3BulkCopyHelper h = new NativeS3BulkCopyHelper(client, 2, 2,
1024);
+ CloseableRegistry registry = new CloseableRegistry();
+
+ assertThatThrownBy(
+ () ->
+ h.copyFiles(
+ java.util.Arrays.asList(
+ CopyRequest.of(
+ new
org.apache.flink.core.fs.Path(
+
"s3://bucket/blocking"),
+ new
org.apache.flink.core.fs.Path(
+
tempDir.resolve("a").toUri()),
+ 1L),
+ CopyRequest.of(
+ new
org.apache.flink.core.fs.Path(
+
"s3://bucket/failing"),
+ new
org.apache.flink.core.fs.Path(
+
tempDir.resolve("b").toUri()),
+ 1L)),
+ registry))
+ .isInstanceOf(IOException.class);
+
+ assertThat(blockingStream.awaitClosed()).isTrue();
+ }
+
+ private static S3AsyncClient asyncClientReturning(
+ CompletableFuture<ResponseInputStream<GetObjectResponse>>
responseFuture) {
+ return asyncClientReturning(responseFuture, null);
+ }
+
+ private static S3AsyncClient asyncClientReturning(
+ CompletableFuture<ResponseInputStream<GetObjectResponse>>
responseFuture,
+ CountDownLatch getObjectCalled) {
+ return (S3AsyncClient)
+ Proxy.newProxyInstance(
+ S3AsyncClient.class.getClassLoader(),
+ new Class<?>[] {S3AsyncClient.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("getObject")
+ && args != null
+ && args.length == 2) {
+ if (getObjectCalled != null) {
+ getObjectCalled.countDown();
+ }
+ return responseFuture;
+ }
+ if (method.getName().equals("close")) {
+ return null;
+ }
+ if (method.getName().equals("serviceName")) {
+ return "s3";
+ }
+ if (method.getName().equals("toString")) {
+ return "test-s3-async-client";
+ }
+ throw new
UnsupportedOperationException(method.toString());
+ });
+ }
+
+ private static final class NonCancellingCompletableFuture<T> extends
CompletableFuture<T> {
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ return false;
+ }
+ }
+
+ private static final class BlockingInputStream extends InputStream {
+ private boolean closed;
+ private boolean readStarted;
+
+ @Override
+ public synchronized int read() throws IOException {
+ byte[] buffer = new byte[1];
+ return read(buffer, 0, 1);
+ }
+
+ @Override
+ public synchronized int read(byte[] b, int off, int len) throws
IOException {
+ readStarted = true;
+ notifyAll();
+ while (!closed) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("interrupted", e);
+ }
+ }
+ throw new IOException("closed");
+ }
+
+ @Override
+ public synchronized void close() {
+ closed = true;
+ notifyAll();
+ }
+
+ synchronized boolean awaitReadStarted() throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (!readStarted && System.nanoTime() < deadline) {
+ TimeUnit.MILLISECONDS.timedWait(this, 10);
+ }
+ return readStarted;
+ }
+
+ synchronized boolean awaitClosed() throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (!closed && System.nanoTime() < deadline) {
+ TimeUnit.MILLISECONDS.timedWait(this, 10);
+ }
+ return closed;
+ }
+ }
}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileIoUtilsTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileIoUtilsTest.java
new file mode 100644
index 00000000000..24fef8d3188
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileIoUtilsTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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 org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.io.ByteArrayInputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link NativeS3FileIoUtils}. */
+class NativeS3FileIoUtilsTest {
+
+ @ParameterizedTest
+ @CsvSource({"1", "7", "1024", "4096"})
+ void testCopyStreamPreservesContentAcrossBufferSizes(int bufferSize,
@TempDir Path tempDir)
+ throws Exception {
+ byte[] data = new byte[3000];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) (i * 31 + 7);
+ }
+ Path dest = tempDir.resolve("out-" + bufferSize + ".bin");
+
+ NativeS3FileIoUtils.copyStream(new ByteArrayInputStream(data), dest,
bufferSize);
+
+ assertThat(Files.readAllBytes(dest)).isEqualTo(data);
+ }
+
+ @Test
+ void testCopyStreamOverwritesExistingFile(@TempDir Path tempDir) throws
Exception {
+ Path dest = tempDir.resolve("out.bin");
+ Files.write(dest, new byte[] {9, 9, 9, 9, 9});
+ byte[] data = {1, 2, 3};
+
+ NativeS3FileIoUtils.copyStream(new ByteArrayInputStream(data), dest,
256 * 1024);
+
+ assertThat(Files.readAllBytes(dest)).isEqualTo(data);
+ }
+
+ @Test
+ void testCopyStreamEmptySource(@TempDir Path tempDir) throws Exception {
+ Path dest = tempDir.resolve("empty.bin");
+
+ NativeS3FileIoUtils.copyStream(new ByteArrayInputStream(new byte[0]),
dest, 1024);
+
+ assertThat(Files.readAllBytes(dest)).isEmpty();
+ }
+
+ @Test
+ void testCreateTemporaryDownloadFilePadsShortNames(@TempDir Path tempDir)
throws Exception {
+ Path shortName = tempDir.resolve("x");
+ Path temp = NativeS3FileIoUtils.createTemporaryDownloadFile(tempDir,
shortName);
+ assertThat(temp).exists();
+ assertThat(temp.getFileName().toString()).endsWith(".tmp");
+ }
+
+ @Test
+ void testMoveFileOverwritesDestination(@TempDir Path tempDir) throws
Exception {
+ Path source = tempDir.resolve("src.bin");
+ Path dest = tempDir.resolve("dst.bin");
+ Files.write(source, new byte[] {1, 2, 3});
+ Files.write(dest, new byte[] {9});
+
+ NativeS3FileIoUtils.moveFile(source, dest);
+
+ assertThat(Files.readAllBytes(dest)).containsExactly(1, 2, 3);
+ assertThat(source).doesNotExist();
+ }
+}
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 f256b0a2040..bc6171919ae 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
@@ -20,19 +20,39 @@ package org.apache.flink.fs.s3native;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.MemorySize;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import java.io.IOException;
import java.net.URI;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link NativeS3FileSystemFactory}. */
class NativeS3FileSystemFactoryTest {
+
+ private final List<NativeS3FileSystem> createdFileSystems = new
ArrayList<>();
+
+ @AfterEach
+ void closeCreatedFileSystems() {
+ for (NativeS3FileSystem fs : createdFileSystems) {
+ try {
+ fs.closeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception ignored) {
+ // best-effort cleanup of test resources
+ }
+ }
+ createdFileSystems.clear();
+ }
+
private static Configuration baseConfig() {
Configuration config = new Configuration();
config.setString("s3.access-key", "test-access-key");
@@ -42,16 +62,22 @@ class NativeS3FileSystemFactoryTest {
return config;
}
- private static NativeS3FileSystem createFs(Configuration config) throws
Exception {
+ private NativeS3FileSystem createFs(Configuration config) throws Exception
{
NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
factory.configure(config);
- return (NativeS3FileSystem)
factory.create(URI.create("s3://test-bucket/"));
+ NativeS3FileSystem fs =
+ (NativeS3FileSystem)
factory.create(URI.create("s3://test-bucket/"));
+ createdFileSystems.add(fs);
+ return fs;
}
- private static NativeS3FileSystem createS3aFs(Configuration config) throws
Exception {
+ private NativeS3FileSystem createS3aFs(Configuration config) throws
Exception {
NativeS3AFileSystemFactory factory = new NativeS3AFileSystemFactory();
factory.configure(config);
- return (NativeS3FileSystem)
factory.create(URI.create("s3a://test-bucket/"));
+ NativeS3FileSystem fs =
+ (NativeS3FileSystem)
factory.create(URI.create("s3a://test-bucket/"));
+ createdFileSystems.add(fs);
+ return fs;
}
@Test
@@ -140,6 +166,14 @@ class NativeS3FileSystemFactoryTest {
assertThat(createFs(config).getClientProvider().isChecksumValidation()).isFalse();
}
+ // --- Bulk copy download buffer ---
+
+ @Test
+ void testBulkCopyDownloadBufferSizeDefaultIs256KB() {
+
assertThat(NativeS3FileSystemFactory.BULK_COPY_DOWNLOAD_BUFFER_SIZE.defaultValue())
+ .isEqualTo(256 * 1024);
+ }
+
// --- Max connections ---
@Test
@@ -164,6 +198,102 @@ class NativeS3FileSystemFactoryTest {
.hasMessageContaining("must be a positive integer");
}
+ @Test
+ void testNonPositiveCrtTargetThroughputThrowsException() {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.CRT_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.CRT_TARGET_THROUGHPUT_GBPS, 0.0);
+ assertThatThrownBy(() -> createFs(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("s3.crt.target-throughput-gbps")
+ .hasMessageContaining("must be positive");
+
+ config.set(NativeS3FileSystemFactory.CRT_TARGET_THROUGHPUT_GBPS, -1.0);
+ assertThatThrownBy(() -> createFs(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("s3.crt.target-throughput-gbps")
+ .hasMessageContaining("must be positive");
+ }
+
+ @Test
+ void testCrtReadAndNativeMemorySettingsArePropagated() throws Exception {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.CRT_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.CRT_READ_BUFFER_SIZE,
MemorySize.ofMebiBytes(1));
+ config.set(
+ NativeS3FileSystemFactory.CRT_MAX_NATIVE_MEMORY_LIMIT,
+ MemorySize.ofMebiBytes(2048));
+
+ S3ClientProvider provider = createFs(config).getClientProvider();
+
+ assertThat(provider.getCrtReadBufferSizeInBytes())
+ .isEqualTo(MemorySize.ofMebiBytes(1).getBytes());
+ assertThat(provider.getCrtMaxNativeMemoryLimitInBytes())
+ .isEqualTo(MemorySize.ofMebiBytes(2048).getBytes());
+ }
+
+ @Test
+ void testCrtReadBufferSizeDefaultsToNullWhenUnset() throws Exception {
+ S3ClientProvider provider = createFs(baseConfig()).getClientProvider();
+ assertThat(provider.getCrtReadBufferSizeInBytes()).isNull();
+ }
+
+ @Test
+ void testCrtMaxConcurrencyDefaultIs256() throws Exception {
+
assertThat(NativeS3FileSystemFactory.CRT_MAX_CONCURRENCY.defaultValue()).isEqualTo(256);
+ S3ClientProvider provider = createFs(baseConfig()).getClientProvider();
+ assertThat(provider.getCrtMaxConcurrency()).isEqualTo(256);
+ }
+
+ @Test
+ void testCrtMaxConcurrencyIsIndependentOfConnectionMax() throws Exception {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.CRT_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.MAX_CONNECTIONS, 50);
+ config.set(NativeS3FileSystemFactory.CRT_MAX_CONCURRENCY, 512);
+
+ S3ClientProvider provider = createFs(config).getClientProvider();
+
+ assertThat(provider.getMaxConnections()).isEqualTo(50);
+ assertThat(provider.getCrtMaxConcurrency()).isEqualTo(512);
+ }
+
+ @Test
+ void testNonPositiveCrtMaxConcurrencyThrowsException() {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.CRT_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.CRT_MAX_CONCURRENCY, 0);
+
+ assertThatThrownBy(() -> createFs(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("s3.crt.max-concurrency")
+ .hasMessageContaining("must be a positive integer");
+ }
+
+ @Test
+ void testNonPositiveCrtReadBufferSizeThrowsException() {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.CRT_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.CRT_READ_BUFFER_SIZE,
MemorySize.ZERO);
+
+ assertThatThrownBy(() -> createFs(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("s3.crt.read-buffer-size")
+ .hasMessageContaining("must be positive");
+ }
+
+ @Test
+ void testNonPositiveCrtMaxNativeMemoryLimitThrowsException() {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.CRT_ENABLED, true);
+ config.set(NativeS3FileSystemFactory.CRT_MAX_NATIVE_MEMORY_LIMIT,
MemorySize.ZERO);
+
+ assertThatThrownBy(() -> createFs(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("s3.crt.max-native-memory-limit")
+ .hasMessageContaining("must be positive");
+ }
+
// --- Max retries ---
@Test
@@ -360,6 +490,27 @@ class NativeS3FileSystemFactoryTest {
assertThat(fs.getBulkCopyHelper().getMaxConcurrentCopies()).isEqualTo(10);
}
+ @Test
+ void testBulkCopyAdvertisesOnlyS3ToLocalCopies() throws Exception {
+ NativeS3FileSystem fs = createFs(baseConfig());
+
+ assertThat(
+ fs.canCopyPaths(
+ new
org.apache.flink.core.fs.Path("s3://test-bucket/state"),
+ new
org.apache.flink.core.fs.Path("file:///tmp/state")))
+ .isTrue();
+ assertThat(
+ fs.canCopyPaths(
+ new
org.apache.flink.core.fs.Path("s3://test-bucket/source"),
+ new
org.apache.flink.core.fs.Path("s3://test-bucket/destination")))
+ .isFalse();
+ assertThat(
+ fs.canCopyPaths(
+ new
org.apache.flink.core.fs.Path("file:///tmp/source"),
+ new
org.apache.flink.core.fs.Path("s3://test-bucket/destination")))
+ .isFalse();
+ }
+
@Test
void testInvalidBulkCopyMaxConcurrentThrowsException() {
Configuration config = baseConfig();
@@ -370,6 +521,16 @@ class NativeS3FileSystemFactoryTest {
.hasMessageContaining("must be a positive integer");
}
+ @Test
+ void testInvalidBulkCopyDownloadBufferSizeThrowsException() {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.BULK_COPY_DOWNLOAD_BUFFER_SIZE,
0);
+ assertThatThrownBy(() -> createFs(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("s3.bulk-copy.download-buffer-size")
+ .hasMessageContaining("must be a positive integer");
+ }
+
// --- Region ---
@Test
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
index 51a457cc537..26985647165 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.fs.s3native;
import
org.apache.flink.fs.s3native.token.DynamicTemporaryAWSCredentialsProvider;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
@@ -31,7 +32,9 @@ import
software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider
import java.lang.reflect.Field;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -42,6 +45,25 @@ class S3ClientProviderTest {
private static final String DUMMY_ENDPOINT = "http://localhost:9000";;
private static final String DUMMY_REGION = "us-east-1";
+ private final List<S3ClientProvider> providers = new ArrayList<>();
+
+ /** Tracks a provider so it is closed after the test, releasing its
SDK/CRT resources. */
+ private S3ClientProvider track(S3ClientProvider provider) {
+ providers.add(provider);
+ return provider;
+ }
+
+ @AfterEach
+ void closeProviders() {
+ for (S3ClientProvider provider : providers) {
+ try {
+ provider.closeAsync().get(10, TimeUnit.SECONDS);
+ } catch (Exception ignored) {
+ }
+ }
+ providers.clear();
+ }
+
@Test
void testMinimalChainWithoutStaticOrCustom() throws Exception {
S3ClientProvider provider =
@@ -271,6 +293,73 @@ class S3ClientProviderTest {
.hasMessageContaining("retryThrottleBaseDelay");
}
+ @Test
+ void testCrtDisabledByDefault() {
+ S3ClientProvider provider =
+ track(
+ S3ClientProvider.builder()
+ .endpoint(DUMMY_ENDPOINT)
+ .region(DUMMY_REGION)
+ .build());
+ assertThat(provider.isUseCrt()).isFalse();
+ // When CRT is disabled the async client must NOT be a CRT-backed
implementation.
+
assertThat(provider.getAsyncClient().getClass().getName()).doesNotContain("Crt");
+ // No Flink-level default applied; getter returns null when user did
not set the value.
+ assertThat(provider.getCrtTargetThroughputGbps()).isNull();
+ assertThat(provider.getCrtReadBufferSizeInBytes()).isNull();
+ assertThat(provider.getCrtMaxNativeMemoryLimitInBytes()).isNull();
+ // CRT max concurrency keeps its builder default (independent of
s3.connection.max).
+ assertThat(provider.getCrtMaxConcurrency())
+
.isEqualTo(NativeS3FileSystemFactory.CRT_MAX_CONCURRENCY.defaultValue());
+ }
+
+ @Test
+ void testCrtFlagIsRecordedAndCrtBranchIsTaken() {
+ S3ClientProvider provider =
+ track(
+ S3ClientProvider.builder()
+ .endpoint(DUMMY_ENDPOINT)
+ .region(DUMMY_REGION)
+ .useCrt(true)
+ .crtTargetThroughputGbps(20.0)
+ .build());
+
+ assertThat(provider.isUseCrt()).isTrue();
+ assertThat(provider.getCrtTargetThroughputGbps()).isEqualTo(20.0);
+
assertThat(provider.getAsyncClient().getClass().getName()).contains("Crt");
+ }
+
+ @Test
+ void testCrtEnabledWithoutThroughputOverrideStillBuildsCrtClient() {
+ S3ClientProvider provider =
+ track(
+ S3ClientProvider.builder()
+ .endpoint(DUMMY_ENDPOINT)
+ .region(DUMMY_REGION)
+ .useCrt(true)
+ .build());
+
+ assertThat(provider.isUseCrt()).isTrue();
+ assertThat(provider.getCrtTargetThroughputGbps()).isNull();
+
assertThat(provider.getAsyncClient().getClass().getName()).contains("Crt");
+ }
+
+ @Test
+ void testCrtMissingJarsMessageIsActionable() {
+ // Contract test: if CRT classes are missing at runtime the user must
get a message
+ // that names the responsible config key, the missing JAR coordinates,
and a setup
+ // pointer. A full classloader-isolation test would require
multi-classloader infra
+ // disproportionate to the value; assert the message contract instead.
+ String msg = S3ClientProvider.Builder.crtMissingJarsMessage();
+ assertThat(msg).contains("s3.crt.enabled=true");
+ // aws-crt-client is now bundled in the fat JAR; only aws-crt (JNI) is
external
+ assertThat(msg).doesNotContain("aws-crt-client");
+ assertThat(msg).contains("aws-crt");
+ assertThat(msg).contains("tools/download-crt-jars.sh");
+ assertThat(msg).contains("plugin");
+ assertThat(msg).contains("README");
+ }
+
@SuppressWarnings("unchecked")
private static List<AwsCredentialsProvider>
extractChain(AwsCredentialsProvider provider)
throws Exception {
diff --git a/flink-filesystems/flink-s3-fs-native/tools/download-crt-jars.sh
b/flink-filesystems/flink-s3-fs-native/tools/download-crt-jars.sh
new file mode 100755
index 00000000000..027816717ef
--- /dev/null
+++ b/flink-filesystems/flink-s3-fs-native/tools/download-crt-jars.sh
@@ -0,0 +1,134 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+# Downloads the aws-crt JAR that flink-s3-fs-native needs when
+# s3.crt.enabled=true.
+#
+# Why only aws-crt and not aws-crt-client:
+# aws-crt-client is pure Java and is bundled (shaded) directly into the
+# flink-s3-fs-native fat JAR at build time — no manual placement needed.
+# aws-crt contains JNI-linked native libraries whose C-side FindClass paths
+# are hardcoded, making Maven shade relocation incompatible. It must
+# therefore be placed in the plugin directory with its original class names.
+#
+# Usage:
+# ./download-crt-jars.sh [OUTPUT_DIR]
+#
+# OUTPUT_DIR Directory to place the JAR in. Defaults to ./crt-jars at the
+# module root. Copy the resulting file to
+# $FLINK_HOME/plugins/s3-fs-native/ alongside
+# flink-s3-fs-native.jar.
+#
+# Environment:
+# AWS_CRT_VERSION Override the auto-resolved aws-crt version (rarely
+# needed; use only when the auto-resolver fails).
+#
+# Requirements: mvn (Apache Maven) on PATH.
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+MODULE_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
+POM_FILE="${MODULE_DIR}/pom.xml"
+OUTPUT_DIR="${1:-${MODULE_DIR}/crt-jars}"
+
+if ! command -v mvn >/dev/null 2>&1; then
+ echo "ERROR: mvn (Apache Maven) is required but not on PATH." >&2
+ exit 1
+fi
+
+if [[ ! -f "${POM_FILE}" ]]; then
+ echo "ERROR: pom.xml not found at ${POM_FILE}." >&2
+ exit 1
+fi
+
+# Read the AWS SDK version that the module compiles against.
+SDK_VERSION="$(sed -n
's|.*<fs.s3.aws.sdk.version>\([^<]*\)</fs.s3.aws.sdk.version>.*|\1|p' \
+ "${POM_FILE}" | head -n1)"
+if [[ -z "${SDK_VERSION}" ]]; then
+ echo "ERROR: could not read <fs.s3.aws.sdk.version> from ${POM_FILE}." >&2
+ exit 1
+fi
+
+# Resolve the aws-crt version. aws-crt uses an independent versioning scheme
+# from the AWS SDK; reading it from the aws-crt-client dependency tree is the
+# authoritative way to stay in sync when the SDK version is bumped.
+resolve_crt_version() {
+ if [[ -n "${AWS_CRT_VERSION:-}" ]]; then
+ echo "${AWS_CRT_VERSION}"
+ return
+ fi
+ local tmp_dir
+ tmp_dir="$(mktemp -d)"
+ # Clean up the probe directory on any return path from this function.
+ trap 'rm -rf "${tmp_dir}"' RETURN
+ local tmp_pom="${tmp_dir}/pom.xml"
+ cat >"${tmp_pom}" <<EOF
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0";>
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.flink.s3native.tools</groupId>
+ <artifactId>crt-version-probe</artifactId>
+ <version>1.0.0</version>
+ <packaging>pom</packaging>
+ <dependencies>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>aws-crt-client</artifactId>
+ <version>${SDK_VERSION}</version>
+ </dependency>
+ </dependencies>
+</project>
+EOF
+ mvn -q -f "${tmp_pom}" dependency:list \
+ -DincludeGroupIds=software.amazon.awssdk.crt \
+ -DincludeArtifactIds=aws-crt \
+ -DexcludeTransitive=false \
+ -DoutputFile=/dev/stdout 2>/dev/null \
+ | awk -F: '/aws-crt/ && $1 ~ /software\.amazon\.awssdk\.crt/
{gsub(/[[:space:]]/, "", $0); print $4; exit}'
+}
+
+CRT_VERSION="$(resolve_crt_version || true)"
+
+if [[ -z "${CRT_VERSION}" ]]; then
+ echo "ERROR: aws-crt version could not be resolved for
aws-crt-client:${SDK_VERSION}." >&2
+ echo " Set AWS_CRT_VERSION explicitly and re-run, e.g." >&2
+ echo " AWS_CRT_VERSION=0.33.6 $0 ${OUTPUT_DIR}" >&2
+ exit 1
+fi
+
+mkdir -p "${OUTPUT_DIR}"
+
+echo "Downloading aws-crt:${CRT_VERSION} into ${OUTPUT_DIR}"
+
+mvn -q dependency:copy \
+ -Dartifact="software.amazon.awssdk.crt:aws-crt:${CRT_VERSION}:jar" \
+ -DoutputDirectory="${OUTPUT_DIR}" \
+ -Dmdep.stripVersion=false
+
+cat <<EOF
+
+Done. Copy the JAR into your Flink plugin directory:
+ cp ${OUTPUT_DIR}/aws-crt-${CRT_VERSION}.jar
\$FLINK_HOME/plugins/s3-fs-native/
+
+Then enable CRT in conf/config.yaml:
+ s3.crt.enabled: true
+
+Note: aws-crt-client is bundled inside flink-s3-fs-native.jar — no separate
+download needed for it.
+EOF
