Skip to content

[FLINK-39110] Add CRT support in native-s3-fs #28136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Samrat002 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-39110] Add CRT support in native-s3-fs #28136

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions flink-filesystems/flink-s3-fs-native/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,18 @@ under the License.
<optional>${flink.markBundledAsOptional}</optional>
</dependency>

<!--
AWS CRT HTTP client (compile-only; NOT bundled in fat JAR due to JNI shading constraints).
Users who enable s3.crt.enabled must place aws-crt-client and aws-crt JARs
in the Flink plugin directory alongside flink-s3-fs-native.jar.
-->
Comment on lines +92 to +96
Copy link
Copy Markdown
Contributor

@gaborgsomogyi gaborgsomogyi May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Some step-by-step would be good on doc page

<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>aws-crt-client</artifactId>
<version>${fs.s3.aws.sdk.version}</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
Expand Down
24 changes: 24 additions & 0 deletions ...nk-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,27 @@ 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 aws-crt-client and aws-crt JARs in the plugin directory "
+ "(they are not bundled in the fat JAR due to JNI shading constraints).");

public static final ConfigOption<Double> CRT_TARGET_THROUGHPUT_GBPS =
Copy link
Copy Markdown
Contributor

@gaborgsomogyi gaborgsomogyi May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this soft or hard limit?

ConfigOptions.key("s3.crt.target-throughput-gbps")
.doubleType()
.defaultValue(10.0)
.withDescription(
"Target throughput in Gbps for the CRT-based S3 async client. "
+ "Only used when s3.crt.enabled is true. "
+ "Higher values allow more concurrent transfers.");

private Configuration flinkConfig;

@Override
Expand Down Expand Up @@ -414,6 +435,9 @@ public FileSystem create(URI fsUri) throws IOException {
.maxRetries(config.get(MAX_RETRIES))
.credentialsProviderClasses(config.get(AWS_CREDENTIALS_PROVIDER))
.encryptionConfig(encryptionConfig)
.useCrt(config.get(CRT_ENABLED))
.crtTargetThroughputGbps(config.get(CRT_TARGET_THROUGHPUT_GBPS))
.crtMinPartSizeInBytes(config.get(PART_UPLOAD_MIN_SIZE))
.build();

NativeS3BulkCopyHelper bulkCopyHelper = null;
Expand Down
112 changes: 92 additions & 20 deletions ...stems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,17 @@
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.core.retry.RetryPolicy;
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;
import software.amazon.awssdk.services.s3.S3AsyncClient;
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.S3CrtRetryConfiguration;
import software.amazon.awssdk.services.sts.StsClient;
import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
Expand Down Expand Up @@ -86,6 +90,8 @@ class S3ClientProvider implements AutoCloseableAsync {
private final boolean checksumValidation;
private final int maxConnections;
private final int maxRetries;
private final boolean useCrt;
private final double crtTargetThroughputGbps;
private final AtomicBoolean closed = new AtomicBoolean(false);

private S3ClientProvider(
Expand All @@ -102,7 +108,9 @@ private S3ClientProvider(
boolean chunkedEncoding,
boolean checksumValidation,
int maxConnections,
int maxRetries) {
int maxRetries,
boolean useCrt,
double crtTargetThroughputGbps) {
this.s3Client = Objects.requireNonNull(s3Client, "s3Client must not be null");
this.transferManager =
Objects.requireNonNull(transferManager, "transferManager must not be null");
Expand All @@ -125,6 +133,8 @@ private S3ClientProvider(
this.checksumValidation = checksumValidation;
this.maxConnections = maxConnections;
this.maxRetries = maxRetries;
this.useCrt = useCrt;
this.crtTargetThroughputGbps = crtTargetThroughputGbps;
}

public S3Client getS3Client() {
Expand Down Expand Up @@ -192,6 +202,16 @@ int getMaxRetries() {
return maxRetries;
}

@VisibleForTesting
boolean isUseCrt() {
return useCrt;
}

@VisibleForTesting
double getCrtTargetThroughputGbps() {
return crtTargetThroughputGbps;
}

@Override
public CompletableFuture<Void> closeAsync() {
if (!closed.compareAndSet(false, true)) {
Expand Down Expand Up @@ -273,6 +293,11 @@ public static class Builder {
// Custom credentials provider class names (comma-separated)
@Nullable private String credentialsProviderClasses;

// CRT configuration
private boolean useCrt = false;
private double crtTargetThroughputGbps = 10.0;
private long crtMinPartSizeInBytes = 5L << 20; // 5MB

public Builder accessKey(@Nullable String accessKey) {
this.accessKey = accessKey;
return this;
Expand Down Expand Up @@ -370,6 +395,21 @@ public Builder credentialsProviderClasses(@Nullable String credentialsProviderCl
return this;
}

public Builder useCrt(boolean useCrt) {
this.useCrt = useCrt;
return this;
}

public Builder crtTargetThroughputGbps(double crtTargetThroughputGbps) {
this.crtTargetThroughputGbps = crtTargetThroughputGbps;
return this;
}

public Builder crtMinPartSizeInBytes(long crtMinPartSizeInBytes) {
this.crtMinPartSizeInBytes = crtMinPartSizeInBytes;
return this;
}

public S3ClientProvider build() {
if (endpoint == null) {
endpoint = System.getProperty("s3.endpoint");
Expand Down Expand Up @@ -418,30 +458,60 @@ public S3ClientProvider build() {
.credentialsProvider(credentialsProvider)
.region(awsRegion)
.serviceConfiguration(s3Config)
.httpClientBuilder(httpClientBuilder)
.overrideConfiguration(overrideConfig);
if (useCrt) {
clientBuilder.httpClientBuilder(
AwsCrtHttpClient.builder()
.maxConcurrency(maxConnections)
.connectionTimeout(connectionTimeout)
.connectionMaxIdleTime(connectionMaxIdleTime));
} else {
clientBuilder.httpClientBuilder(httpClientBuilder);
}
if (endpointUri != null) {
clientBuilder.endpointOverride(endpointUri);
}
S3Client s3Client = clientBuilder.build();

S3AsyncClient asyncClient;
if (useCrt) {
S3CrtAsyncClientBuilder crtAsyncBuilder =
S3AsyncClient.crtBuilder()
.credentialsProvider(credentialsProvider)
.region(awsRegion)
.forcePathStyle(pathStyleAccess)
.checksumValidationEnabled(checksumValidation)
.retryConfiguration(
S3CrtRetryConfiguration.builder()
.numRetries(maxRetries)
.build())
.maxConcurrency(maxConnections)
.targetThroughputInGbps((double) crtTargetThroughputGbps)
.minimumPartSizeInBytes(crtMinPartSizeInBytes);
if (endpointUri != null) {
crtAsyncBuilder.endpointOverride(endpointUri);
}
asyncClient = crtAsyncBuilder.build();
} else {
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);
}
asyncClient = asyncBuilder.build();
}
S3TransferManager transferManager =
S3TransferManager.builder()
.s3Client(
S3AsyncClient.builder()
.credentialsProvider(credentialsProvider)
.region(awsRegion)
.serviceConfiguration(s3Config)
.httpClientBuilder(
NettyNioAsyncHttpClient.builder()
.maxConcurrency(maxConnections)
.connectionTimeout(connectionTimeout)
.readTimeout(socketTimeout)
.connectionAcquisitionTimeout(
connectionTimeout))
.overrideConfiguration(overrideConfig)
.endpointOverride(endpointUri)
.build())
.build();
S3TransferManager.builder().s3Client(asyncClient).build();

return new S3ClientProvider(
s3Client,
Expand All @@ -457,7 +527,9 @@ public S3ClientProvider build() {
chunkedEncoding,
checksumValidation,
maxConnections,
maxRetries);
maxRetries,
useCrt,
crtTargetThroughputGbps);
}

private AwsCredentialsProvider buildBaseCredentialsProvider() {
Expand Down
21 changes: 21 additions & 0 deletions ...s/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,27 @@ void testEmptyProviderStringThrows() {
.hasMessageContaining("no valid provider class names");
}

@Test
void testCrtDisabledByDefault() {
S3ClientProvider provider =
S3ClientProvider.builder().endpoint(DUMMY_ENDPOINT).region(DUMMY_REGION).build();
assertThat(provider.isUseCrt()).isFalse();
}

@Test
void testCrtFlagIsRecorded() {
S3ClientProvider provider =
S3ClientProvider.builder()
.endpoint(DUMMY_ENDPOINT)
.region(DUMMY_REGION)
.useCrt(true)
.crtTargetThroughputGbps(20.0)
.build();

assertThat(provider.isUseCrt()).isTrue();
assertThat(provider.getCrtTargetThroughputGbps()).isEqualTo(20.0);
}

@SuppressWarnings("unchecked")
private static List<AwsCredentialsProvider> extractChain(AwsCredentialsProvider provider)
throws Exception {
Expand Down