You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by am...@apache.org on 2025/05/22 16:11:51 UTC
(iceberg) branch main updated: AWS: Support multiple storage credential prefixes (#12799)
This is an automated email from the ASF dual-hosted git repository.
amoghj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 14493784a6 AWS: Support multiple storage credential prefixes (#12799)
14493784a6 is described below
commit 14493784a6d59d7d1dfee84fe619fe349518e537
Author: Eduard Tudenhoefner <et...@gmail.com>
AuthorDate: Thu May 22 18:11:45 2025 +0200
AWS: Support multiple storage credential prefixes (#12799)
---
.../org/apache/iceberg/aws/s3/TestS3FileIO.java | 227 +++++++++++++++++----
.../apache/iceberg/aws/s3/PrefixedS3Client.java | 121 +++++++++++
.../java/org/apache/iceberg/aws/s3/S3FileIO.java | 210 ++++++++++---------
.../apache/iceberg/aws/s3/S3FileIOProperties.java | 4 +
.../org/apache/iceberg/aws/s3/S3InputFile.java | 44 ++++
.../org/apache/iceberg/aws/s3/S3OutputFile.java | 24 +++
.../iceberg/aws/s3/TestPrefixedS3Client.java | 60 ++++++
7 files changed, 547 insertions(+), 143 deletions(-)
diff --git a/aws/src/integration/java/org/apache/iceberg/aws/s3/TestS3FileIO.java b/aws/src/integration/java/org/apache/iceberg/aws/s3/TestS3FileIO.java
index af5091721b..85453cba97 100644
--- a/aws/src/integration/java/org/apache/iceberg/aws/s3/TestS3FileIO.java
+++ b/aws/src/integration/java/org/apache/iceberg/aws/s3/TestS3FileIO.java
@@ -57,6 +57,7 @@ import org.apache.iceberg.ManifestWriter;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.TestHelpers;
+import org.apache.iceberg.aws.AwsClientProperties;
import org.apache.iceberg.aws.AwsProperties;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.common.DynMethods;
@@ -91,6 +92,7 @@ import org.testcontainers.junit.jupiter.Testcontainers;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.BucketAlreadyExistsException;
import software.amazon.awssdk.services.s3.model.BucketAlreadyOwnedByYouException;
@@ -483,6 +485,55 @@ public class TestS3FileIO {
.isEqualTo(fileIO.credentials());
}
+ @Test
+ public void fileIOWithPrefixedS3ClientKryoSerialization() throws IOException {
+ S3FileIO io = new S3FileIO();
+ io.setCredentials(
+ ImmutableList.of(
+ StorageCredential.create("s3://my-bucket/my-path/table1", Map.of("key1", "val1"))));
+ io.initialize(Map.of(AwsClientProperties.CLIENT_REGION, "us-east-1"));
+
+ // there should be a client for the generic and specific storage prefix available
+ assertThat(io.client()).isInstanceOf(S3Client.class);
+ assertThat(io.asyncClient()).isInstanceOf(S3AsyncClient.class);
+ assertThat(io.client("s3://my-bucket/my-path")).isInstanceOf(S3Client.class);
+ assertThat(io.asyncClient("s3://my-bucket/my-path")).isInstanceOf(S3AsyncClient.class);
+
+ S3FileIO fileIO = TestHelpers.KryoHelpers.roundTripSerialize(io);
+ assertThat(fileIO.credentials()).isEqualTo(io.credentials());
+
+ // make sure there's a client for the generic and specific storage prefix available after ser/de
+ assertThat(fileIO.client()).isInstanceOf(S3Client.class);
+ assertThat(fileIO.asyncClient()).isInstanceOf(S3AsyncClient.class);
+ assertThat(fileIO.client("s3://my-bucket/my-path")).isInstanceOf(S3Client.class);
+ assertThat(fileIO.asyncClient("s3://my-bucket/my-path")).isInstanceOf(S3AsyncClient.class);
+ }
+
+ @Test
+ public void fileIOWithPrefixedS3ClientJavaSerialization()
+ throws IOException, ClassNotFoundException {
+ S3FileIO io = new S3FileIO();
+ io.setCredentials(
+ ImmutableList.of(
+ StorageCredential.create("s3://my-bucket/my-path/table1", Map.of("key1", "val1"))));
+ io.initialize(Map.of(AwsClientProperties.CLIENT_REGION, "us-east-1"));
+
+ // there should be a client for the generic and specific storage prefix available
+ assertThat(io.client()).isInstanceOf(S3Client.class);
+ assertThat(io.asyncClient()).isInstanceOf(S3AsyncClient.class);
+ assertThat(io.client("s3://my-bucket/my-path")).isInstanceOf(S3Client.class);
+ assertThat(io.asyncClient("s3://my-bucket/my-path")).isInstanceOf(S3AsyncClient.class);
+
+ S3FileIO fileIO = TestHelpers.roundTripSerialize(io);
+ assertThat(fileIO.credentials()).isEqualTo(io.credentials());
+
+ // make sure there's a client for the generic and specific storage prefix available after ser/de
+ assertThat(fileIO.client()).isInstanceOf(S3Client.class);
+ assertThat(fileIO.asyncClient()).isInstanceOf(S3AsyncClient.class);
+ assertThat(fileIO.client("s3://my-bucket/my-path")).isInstanceOf(S3Client.class);
+ assertThat(fileIO.asyncClient("s3://my-bucket/my-path")).isInstanceOf(S3AsyncClient.class);
+ }
+
@Test
public void testS3FileIOJavaSerialization() throws IOException, ClassNotFoundException {
FileIO testS3FileIO = new S3FileIO();
@@ -570,50 +621,83 @@ public class TestS3FileIO {
@Test
public void resolvingFileIOLoadWithStorageCredentials()
throws IOException, ClassNotFoundException {
- StorageCredential credential = StorageCredential.create("prefix", Map.of("key1", "val1"));
+ StorageCredential credential = StorageCredential.create("s3://foo/bar", Map.of("key1", "val1"));
List<StorageCredential> storageCredentials = ImmutableList.of(credential);
ResolvingFileIO resolvingFileIO = new ResolvingFileIO();
resolvingFileIO.setCredentials(storageCredentials);
- resolvingFileIO.initialize(ImmutableMap.of());
+ resolvingFileIO.initialize(ImmutableMap.of(AwsClientProperties.CLIENT_REGION, "us-east-1"));
FileIO result =
DynMethods.builder("io")
.hiddenImpl(ResolvingFileIO.class, String.class)
.build(resolvingFileIO)
.invoke("s3://foo/bar");
- assertThat(result)
- .isInstanceOf(S3FileIO.class)
- .asInstanceOf(InstanceOfAssertFactories.type(S3FileIO.class))
- .extracting(S3FileIO::credentials)
- .isEqualTo(storageCredentials);
+ ObjectAssert<S3FileIO> io =
+ assertThat(result)
+ .isInstanceOf(S3FileIO.class)
+ .asInstanceOf(InstanceOfAssertFactories.type(S3FileIO.class));
+ io.extracting(S3FileIO::credentials).isEqualTo(storageCredentials);
+ io.satisfies(
+ fileIO -> {
+ // make sure there are two separate S3 clients for different prefixes and that the
+ // underlying sync/async client is set
+ assertThat(fileIO.client("s3://foo/bar"))
+ .isNotSameAs(fileIO.client())
+ .isInstanceOf(S3Client.class);
+ assertThat(fileIO.asyncClient("s3://foo/bar"))
+ .isNotSameAs(fileIO.asyncClient())
+ .isInstanceOf(S3AsyncClient.class);
+ });
// make sure credentials are still present after kryo serde
- ResolvingFileIO io = TestHelpers.KryoHelpers.roundTripSerialize(resolvingFileIO);
- assertThat(io.credentials()).isEqualTo(storageCredentials);
+ ResolvingFileIO resolvingIO = TestHelpers.KryoHelpers.roundTripSerialize(resolvingFileIO);
+ assertThat(resolvingIO.credentials()).isEqualTo(storageCredentials);
result =
DynMethods.builder("io")
.hiddenImpl(ResolvingFileIO.class, String.class)
- .build(io)
+ .build(resolvingIO)
.invoke("s3://foo/bar");
- assertThat(result)
- .isInstanceOf(S3FileIO.class)
- .asInstanceOf(InstanceOfAssertFactories.type(S3FileIO.class))
- .extracting(S3FileIO::credentials)
- .isEqualTo(storageCredentials);
+ io =
+ assertThat(result)
+ .isInstanceOf(S3FileIO.class)
+ .asInstanceOf(InstanceOfAssertFactories.type(S3FileIO.class));
+ io.extracting(S3FileIO::credentials).isEqualTo(storageCredentials);
+ io.satisfies(
+ fileIO -> {
+ // make sure there are two separate S3 clients for different prefixes and that the
+ // underlying sync/async client is set
+ assertThat(fileIO.client("s3://foo/bar"))
+ .isNotSameAs(fileIO.client())
+ .isInstanceOf(S3Client.class);
+ assertThat(fileIO.asyncClient("s3://foo/bar"))
+ .isNotSameAs(fileIO.asyncClient())
+ .isInstanceOf(S3AsyncClient.class);
+ });
// make sure credentials are still present after java serde
- io = TestHelpers.roundTripSerialize(resolvingFileIO);
- assertThat(io.credentials()).isEqualTo(storageCredentials);
+ resolvingIO = TestHelpers.roundTripSerialize(resolvingFileIO);
+ assertThat(resolvingIO.credentials()).isEqualTo(storageCredentials);
result =
DynMethods.builder("io")
.hiddenImpl(ResolvingFileIO.class, String.class)
- .build(io)
+ .build(resolvingIO)
.invoke("s3://foo/bar");
- assertThat(result)
- .isInstanceOf(S3FileIO.class)
- .asInstanceOf(InstanceOfAssertFactories.type(S3FileIO.class))
- .extracting(S3FileIO::credentials)
- .isEqualTo(storageCredentials);
+ io =
+ assertThat(result)
+ .isInstanceOf(S3FileIO.class)
+ .asInstanceOf(InstanceOfAssertFactories.type(S3FileIO.class));
+ io.extracting(S3FileIO::credentials).isEqualTo(storageCredentials);
+ io.satisfies(
+ fileIO -> {
+ // make sure there are two separate S3 clients for different prefixes and that the
+ // underlying sync/async client is set
+ assertThat(fileIO.client("s3://foo/bar"))
+ .isNotSameAs(fileIO.client())
+ .isInstanceOf(S3Client.class);
+ assertThat(fileIO.asyncClient("s3://foo/bar"))
+ .isNotSameAs(fileIO.asyncClient())
+ .isInstanceOf(S3AsyncClient.class);
+ });
}
@Test
@@ -622,6 +706,8 @@ public class TestS3FileIO {
fileIO.setCredentials(ImmutableList.of());
fileIO.initialize(
ImmutableMap.of(
+ AwsClientProperties.CLIENT_REGION,
+ "us-east-1",
"s3.access-key-id",
"keyIdFromProperties",
"s3.secret-access-key",
@@ -629,10 +715,14 @@ public class TestS3FileIO {
"s3.session-token",
"sessionTokenFromProperties"));
+ // make sure that the generic S3 Client is used for all storage paths if there are no storage
+ // credentials configured
+ assertThat(fileIO.client("s3://my-bucket/table1"))
+ .isSameAs(fileIO.client("s3://random-bucket/"))
+ .isSameAs(fileIO.client("s3a://random-bucket/tableX"));
+
ObjectAssert<S3FileIOProperties> s3FileIOProperties =
- assertThat(fileIO)
- .extracting("s3FileIOProperties")
- .asInstanceOf(InstanceOfAssertFactories.type(S3FileIOProperties.class));
+ assertThat(fileIO.clientForStoragePath("s3://my-bucket/table1").s3FileIOProperties());
s3FileIOProperties.extracting(S3FileIOProperties::accessKeyId).isEqualTo("keyIdFromProperties");
s3FileIOProperties
.extracting(S3FileIOProperties::secretAccessKey)
@@ -659,6 +749,8 @@ public class TestS3FileIO {
fileIO.setCredentials(ImmutableList.of(s3Credential));
fileIO.initialize(
ImmutableMap.of(
+ AwsClientProperties.CLIENT_REGION,
+ "us-east-1",
"s3.access-key-id",
"keyIdFromProperties",
"s3.secret-access-key",
@@ -666,10 +758,23 @@ public class TestS3FileIO {
"s3.session-token",
"sessionTokenFromProperties"));
+ // verify that the generic S3 client is used if the storage prefix doesn't match the prefixes in
+ // the storage credentials
+ assertThat(fileIO.client("s3://my-bucket/table1"))
+ .isNotSameAs(fileIO.client("s3://custom-uri/table2"));
+ ObjectAssert<S3FileIOProperties> genericProperties =
+ assertThat(fileIO.clientForStoragePath("s3://my-bucket/table1").s3FileIOProperties());
+ genericProperties.extracting(S3FileIOProperties::accessKeyId).isEqualTo("keyIdFromProperties");
+ genericProperties
+ .extracting(S3FileIOProperties::secretAccessKey)
+ .isEqualTo("accessKeyFromProperties");
+ genericProperties
+ .extracting(S3FileIOProperties::sessionToken)
+ .isEqualTo("sessionTokenFromProperties");
+
ObjectAssert<S3FileIOProperties> s3FileIOProperties =
- assertThat(fileIO)
- .extracting("s3FileIOProperties")
- .asInstanceOf(InstanceOfAssertFactories.type(S3FileIOProperties.class));
+ assertThat(
+ fileIO.clientForStoragePath("s3://custom-uri/random/table1").s3FileIOProperties());
s3FileIOProperties.extracting(S3FileIOProperties::accessKeyId).isEqualTo("keyIdFromCredential");
s3FileIOProperties
.extracting(S3FileIOProperties::secretAccessKey)
@@ -705,18 +810,58 @@ public class TestS3FileIO {
S3FileIO fileIO = new S3FileIO();
fileIO.setCredentials(ImmutableList.of(s3Credential1, s3Credential2));
- assertThatThrownBy(
- () ->
- fileIO.initialize(
- ImmutableMap.of(
- "s3.access-key-id",
- "keyIdFromProperties",
- "s3.secret-access-key",
- "accessKeyFromProperties",
- "s3.session-token",
- "sessionTokenFromProperties")))
- .isInstanceOf(IllegalStateException.class)
- .hasMessage("Invalid S3 Credentials: only one S3 credential should exist");
+ fileIO.initialize(
+ ImmutableMap.of(
+ AwsClientProperties.CLIENT_REGION,
+ "us-east-1",
+ "s3.access-key-id",
+ "keyIdFromProperties",
+ "s3.secret-access-key",
+ "accessKeyFromProperties",
+ "s3.session-token",
+ "sessionTokenFromProperties"));
+
+ // verify that the generic S3 client is used if the storage prefix doesn't match the prefixes in
+ // the storage credentials
+ assertThat(fileIO.client("s3://my-bucket/table1"))
+ .isNotSameAs(fileIO.client("s3://custom-uri/1"))
+ .isNotSameAs(fileIO.client("s3://custom-uri/2"));
+ ObjectAssert<S3FileIOProperties> genericProperties =
+ assertThat(fileIO.clientForStoragePath("s3://my-bucket/table1").s3FileIOProperties());
+ genericProperties.extracting(S3FileIOProperties::accessKeyId).isEqualTo("keyIdFromProperties");
+ genericProperties
+ .extracting(S3FileIOProperties::secretAccessKey)
+ .isEqualTo("accessKeyFromProperties");
+ genericProperties
+ .extracting(S3FileIOProperties::sessionToken)
+ .isEqualTo("sessionTokenFromProperties");
+
+ // there are separate credentials configured for each storage path, so there should be separate
+ // S3 client instances
+ assertThat(fileIO.client("s3://custom-uri/1")).isNotSameAs(fileIO.client("s3://custom-uri/2"));
+ ObjectAssert<S3FileIOProperties> s3FileIOProperties1 =
+ assertThat(fileIO.clientForStoragePath("s3://custom-uri/1/table1").s3FileIOProperties());
+ s3FileIOProperties1
+ .extracting(S3FileIOProperties::accessKeyId)
+ .isEqualTo("keyIdFromCredential1");
+ s3FileIOProperties1
+ .extracting(S3FileIOProperties::secretAccessKey)
+ .isEqualTo("accessKeyFromCredential1");
+ s3FileIOProperties1
+ .extracting(S3FileIOProperties::sessionToken)
+ .isEqualTo("sessionTokenFromCredential1");
+
+ ObjectAssert<S3FileIOProperties> s3FileIOProperties2 =
+ assertThat(fileIO.clientForStoragePath("s3://custom-uri/2/table1").s3FileIOProperties());
+ s3FileIOProperties2
+ .extracting(S3FileIOProperties::accessKeyId)
+ .isEqualTo("keyIdFromCredential2");
+ s3FileIOProperties2
+ .extracting(S3FileIOProperties::secretAccessKey)
+ .isEqualTo("accessKeyFromCredential2");
+ s3FileIOProperties2
+ .extracting(S3FileIOProperties::sessionToken)
+ .isEqualTo("sessionTokenFromCredential2");
}
private void createRandomObjects(String prefix, int count) {
diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/PrefixedS3Client.java b/aws/src/main/java/org/apache/iceberg/aws/s3/PrefixedS3Client.java
new file mode 100644
index 0000000000..b0b65ab8ea
--- /dev/null
+++ b/aws/src/main/java/org/apache/iceberg/aws/s3/PrefixedS3Client.java
@@ -0,0 +1,121 @@
+/*
+ * 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.iceberg.aws.s3;
+
+import java.util.Map;
+import org.apache.iceberg.aws.AwsClientFactory;
+import org.apache.iceberg.aws.S3FileIOAwsClientFactories;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.SerializableSupplier;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Client;
+
+class PrefixedS3Client implements AutoCloseable {
+
+ private final String storagePrefix;
+ private final S3FileIOProperties s3FileIOProperties;
+ private SerializableSupplier<S3Client> s3;
+ private SerializableSupplier<S3AsyncClient> s3Async;
+ private transient volatile S3Client s3Client;
+ private transient volatile S3AsyncClient s3AsyncClient;
+
+ PrefixedS3Client(
+ String storagePrefix,
+ Map<String, String> properties,
+ SerializableSupplier<S3Client> s3,
+ SerializableSupplier<S3AsyncClient> s3Async) {
+ Preconditions.checkArgument(
+ !Strings.isNullOrEmpty(storagePrefix), "Invalid storage prefix: null or empty");
+ Preconditions.checkArgument(null != properties, "Invalid properties: null");
+ this.storagePrefix = storagePrefix;
+ this.s3 = s3;
+ this.s3Async = s3Async;
+ this.s3FileIOProperties = new S3FileIOProperties(properties);
+ // Do not override s3 client if it was provided
+ if (s3 == null) {
+ Object clientFactory = S3FileIOAwsClientFactories.initialize(properties);
+ if (clientFactory instanceof S3FileIOAwsClientFactory) {
+ this.s3 = ((S3FileIOAwsClientFactory) clientFactory)::s3;
+ }
+ if (clientFactory instanceof AwsClientFactory) {
+ this.s3 = ((AwsClientFactory) clientFactory)::s3;
+ }
+ if (s3FileIOProperties.isPreloadClientEnabled()) {
+ s3();
+ }
+ }
+
+ // Do not override s3Async client if it was provided
+ if (s3Async == null) {
+ Object clientFactory = S3FileIOAwsClientFactories.initialize(properties);
+ if (clientFactory instanceof S3FileIOAwsClientFactory) {
+ this.s3Async = ((S3FileIOAwsClientFactory) clientFactory)::s3Async;
+ }
+ if (clientFactory instanceof AwsClientFactory) {
+ this.s3Async = ((AwsClientFactory) clientFactory)::s3Async;
+ }
+ }
+ }
+
+ public String storagePrefix() {
+ return storagePrefix;
+ }
+
+ public S3Client s3() {
+ if (s3Client == null) {
+ synchronized (this) {
+ if (s3Client == null) {
+ s3Client = s3.get();
+ }
+ }
+ }
+
+ return s3Client;
+ }
+
+ public S3AsyncClient s3Async() {
+ if (s3AsyncClient == null) {
+ synchronized (this) {
+ if (s3AsyncClient == null) {
+ s3AsyncClient = s3Async.get();
+ }
+ }
+ }
+
+ return s3AsyncClient;
+ }
+
+ public S3FileIOProperties s3FileIOProperties() {
+ return s3FileIOProperties;
+ }
+
+ @Override
+ public void close() {
+ if (null != s3Client) {
+ s3Client.close();
+ }
+
+ if (null != s3AsyncClient) {
+ // cleanup usage in analytics accelerator if any
+ AnalyticsAcceleratorUtil.cleanupCache(s3AsyncClient, s3FileIOProperties);
+ s3AsyncClient.close();
+ }
+ }
+}
diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
index 2838980d4c..43a09f4654 100644
--- a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
+++ b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
@@ -30,7 +30,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
-import org.apache.iceberg.aws.AwsClientFactory;
import org.apache.iceberg.aws.S3FileIOAwsClientFactories;
import org.apache.iceberg.common.DynConstructors;
import org.apache.iceberg.io.BulkDeletionFailureException;
@@ -43,6 +42,7 @@ import org.apache.iceberg.io.StorageCredential;
import org.apache.iceberg.io.SupportsRecoveryOperations;
import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Joiner;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -93,19 +93,18 @@ public class S3FileIO
private static final Logger LOG = LoggerFactory.getLogger(S3FileIO.class);
private static final String DEFAULT_METRICS_IMPL =
"org.apache.iceberg.hadoop.HadoopMetricsContext";
+ private static final String ROOT_PREFIX = "s3";
private static volatile ExecutorService executorService;
private String credential = null;
private SerializableSupplier<S3Client> s3;
private SerializableSupplier<S3AsyncClient> s3Async;
- private S3FileIOProperties s3FileIOProperties;
private SerializableMap<String, String> properties = null;
- private transient volatile S3Client client;
- private transient volatile S3AsyncClient asyncClient;
private MetricsContext metrics = MetricsContext.nullMetrics();
private final AtomicBoolean isResourceClosed = new AtomicBoolean(false);
private transient StackTraceElement[] createStack;
private List<StorageCredential> storageCredentials = ImmutableList.of();
+ private transient volatile Map<String, PrefixedS3Client> clientByPrefix;
/**
* No-arg constructor to load the FileIO dynamically.
@@ -171,40 +170,32 @@ public class S3FileIO
S3FileIOProperties s3FileIOProperties) {
this.s3 = s3;
this.s3Async = s3Async;
- this.s3FileIOProperties = s3FileIOProperties;
this.createStack = Thread.currentThread().getStackTrace();
+ this.properties = SerializableMap.copyOf(s3FileIOProperties.properties());
}
@Override
public InputFile newInputFile(String path) {
- if (shouldUseAsyncClient()) {
- return S3InputFile.fromLocation(path, client(), asyncClient(), s3FileIOProperties, metrics);
- }
- return S3InputFile.fromLocation(path, client(), s3FileIOProperties, metrics);
+ return S3InputFile.fromLocation(path, clientForStoragePath(path), metrics);
}
@Override
public InputFile newInputFile(String path, long length) {
- if (shouldUseAsyncClient()) {
- return S3InputFile.fromLocation(
- path, length, client(), asyncClient(), s3FileIOProperties, metrics);
- }
- return S3InputFile.fromLocation(path, length, client(), s3FileIOProperties, metrics);
+ return S3InputFile.fromLocation(path, length, clientForStoragePath(path), metrics);
}
@Override
public OutputFile newOutputFile(String path) {
- if (shouldUseAsyncClient()) {
- return S3OutputFile.fromLocation(path, client(), asyncClient(), s3FileIOProperties, metrics);
- }
- return S3OutputFile.fromLocation(path, client(), s3FileIOProperties, metrics);
+ return S3OutputFile.fromLocation(path, clientForStoragePath(path), metrics);
}
@Override
public void deleteFile(String path) {
+ PrefixedS3Client client = clientForStoragePath(path);
+ S3FileIOProperties s3FileIOProperties = client.s3FileIOProperties();
if (s3FileIOProperties.deleteTags() != null && !s3FileIOProperties.deleteTags().isEmpty()) {
try {
- tagFileToDelete(path, s3FileIOProperties.deleteTags());
+ tagFileToDelete(client, path, s3FileIOProperties.deleteTags());
} catch (S3Exception e) {
LOG.warn("Failed to add delete tags: {} to {}", s3FileIOProperties.deleteTags(), path, e);
}
@@ -218,7 +209,7 @@ public class S3FileIO
DeleteObjectRequest deleteRequest =
DeleteObjectRequest.builder().bucket(location.bucket()).key(location.key()).build();
- client().deleteObject(deleteRequest);
+ client.s3().deleteObject(deleteRequest);
}
@Override
@@ -236,6 +227,7 @@ public class S3FileIO
*/
@Override
public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException {
+ S3FileIOProperties s3FileIOProperties = clientForStoragePath(ROOT_PREFIX).s3FileIOProperties();
if (s3FileIOProperties.deleteTags() != null && !s3FileIOProperties.deleteTags().isEmpty()) {
Tasks.foreach(paths)
.noRetry()
@@ -248,7 +240,10 @@ public class S3FileIO
s3FileIOProperties.deleteTags(),
path,
exc))
- .run(path -> tagFileToDelete(path, s3FileIOProperties.deleteTags()));
+ .run(
+ path ->
+ tagFileToDelete(
+ clientForStoragePath(path), path, s3FileIOProperties.deleteTags()));
}
if (s3FileIOProperties.isDeleteEnabled()) {
@@ -256,14 +251,15 @@ public class S3FileIO
Multimaps.newSetMultimap(Maps.newHashMap(), Sets::newHashSet);
List<Future<List<String>>> deletionTasks = Lists.newArrayList();
for (String path : paths) {
- S3URI location = new S3URI(path, s3FileIOProperties.bucketToAccessPointMapping());
+ PrefixedS3Client client = clientForStoragePath(path);
+ S3URI location = new S3URI(path, client.s3FileIOProperties().bucketToAccessPointMapping());
String bucket = location.bucket();
String objectKey = location.key();
bucketToObjects.get(bucket).add(objectKey);
- if (bucketToObjects.get(bucket).size() == s3FileIOProperties.deleteBatchSize()) {
+ if (bucketToObjects.get(bucket).size() == client.s3FileIOProperties().deleteBatchSize()) {
Set<String> keys = Sets.newHashSet(bucketToObjects.get(bucket));
Future<List<String>> deletionTask =
- executorService().submit(() -> deleteBatch(bucket, keys));
+ executorService().submit(() -> deleteBatch(client, bucket, keys));
deletionTasks.add(deletionTask);
bucketToObjects.removeAll(bucket);
}
@@ -275,7 +271,8 @@ public class S3FileIO
String bucket = bucketToObjectsEntry.getKey();
Collection<String> keys = bucketToObjectsEntry.getValue();
Future<List<String>> deletionTask =
- executorService().submit(() -> deleteBatch(bucket, keys));
+ executorService()
+ .submit(() -> deleteBatch(clientForStoragePath("s3://" + bucket), bucket, keys));
deletionTasks.add(deletionTask);
}
@@ -301,14 +298,15 @@ public class S3FileIO
}
}
- private void tagFileToDelete(String path, Set<Tag> deleteTags) throws S3Exception {
- S3URI location = new S3URI(path, s3FileIOProperties.bucketToAccessPointMapping());
+ private void tagFileToDelete(PrefixedS3Client client, String path, Set<Tag> deleteTags)
+ throws S3Exception {
+ S3URI location = new S3URI(path, client.s3FileIOProperties().bucketToAccessPointMapping());
String bucket = location.bucket();
String objectKey = location.key();
GetObjectTaggingRequest getObjectTaggingRequest =
GetObjectTaggingRequest.builder().bucket(bucket).key(objectKey).build();
GetObjectTaggingResponse getObjectTaggingResponse =
- client().getObjectTagging(getObjectTaggingRequest);
+ client.s3().getObjectTagging(getObjectTaggingRequest);
// Get existing tags, if any and then add the delete tags
Set<Tag> tags = Sets.newHashSet();
if (getObjectTaggingResponse.hasTagSet()) {
@@ -322,10 +320,11 @@ public class S3FileIO
.key(objectKey)
.tagging(Tagging.builder().tagSet(tags).build())
.build();
- client().putObjectTagging(putObjectTaggingRequest);
+ client.s3().putObjectTagging(putObjectTaggingRequest);
}
- private List<String> deleteBatch(String bucket, Collection<String> keysToDelete) {
+ private List<String> deleteBatch(
+ PrefixedS3Client client, String bucket, Collection<String> keysToDelete) {
List<ObjectIdentifier> objectIds =
keysToDelete.stream()
.map(key -> ObjectIdentifier.builder().key(key).build())
@@ -337,7 +336,7 @@ public class S3FileIO
.build();
List<String> failures = Lists.newArrayList();
try {
- DeleteObjectsResponse response = client().deleteObjects(request);
+ DeleteObjectsResponse response = client.s3().deleteObjects(request);
if (response.hasErrors()) {
failures.addAll(
response.errors().stream()
@@ -356,9 +355,11 @@ public class S3FileIO
@Override
public Iterable<FileInfo> listPrefix(String prefix) {
- S3URI uri = new S3URI(prefix, s3FileIOProperties.bucketToAccessPointMapping());
+ PrefixedS3Client client = clientForStoragePath(prefix);
+
+ S3URI uri = new S3URI(prefix, client.s3FileIOProperties().bucketToAccessPointMapping());
if (uri.useS3DirectoryBucket()
- && s3FileIOProperties.isS3DirectoryBucketListPrefixAsDirectory()) {
+ && client.s3FileIOProperties().isS3DirectoryBucketListPrefixAsDirectory()) {
uri = uri.toDirectoryPath();
}
@@ -367,7 +368,7 @@ public class S3FileIO
ListObjectsV2Request.builder().bucket(s3uri.bucket()).prefix(s3uri.key()).build();
return () ->
- client().listObjectsV2Paginator(request).stream()
+ client.s3().listObjectsV2Paginator(request).stream()
.flatMap(r -> r.contents().stream())
.map(
o ->
@@ -392,29 +393,71 @@ public class S3FileIO
}
public S3Client client() {
- if (client == null) {
- synchronized (this) {
- if (client == null) {
- client = s3.get();
- }
+ return client(ROOT_PREFIX);
+ }
+
+ @SuppressWarnings("resource")
+ public S3Client client(String storagePath) {
+ return clientForStoragePath(storagePath).s3();
+ }
+
+ public S3AsyncClient asyncClient() {
+ return asyncClient(ROOT_PREFIX);
+ }
+
+ @SuppressWarnings("resource")
+ public S3AsyncClient asyncClient(String storagePath) {
+ return clientForStoragePath(storagePath).s3Async();
+ }
+
+ @VisibleForTesting
+ PrefixedS3Client clientForStoragePath(String storagePath) {
+ PrefixedS3Client client;
+ String matchingPrefix = ROOT_PREFIX;
+
+ for (String storagePrefix : clientByPrefix().keySet()) {
+ if (storagePath.startsWith(storagePrefix)
+ && storagePrefix.length() > matchingPrefix.length()) {
+ matchingPrefix = storagePrefix;
}
}
+
+ client = clientByPrefix().getOrDefault(matchingPrefix, null);
+
+ Preconditions.checkState(
+ null != client, "[BUG] S3 client for storage path not available: %s", storagePath);
return client;
}
- public S3AsyncClient asyncClient() {
- if (asyncClient == null) {
+ private Map<String, PrefixedS3Client> clientByPrefix() {
+ if (null == clientByPrefix) {
synchronized (this) {
- if (asyncClient == null) {
- asyncClient = s3Async.get();
+ if (null == clientByPrefix) {
+ this.clientByPrefix = Maps.newHashMap();
+
+ clientByPrefix.put(
+ ROOT_PREFIX, new PrefixedS3Client(ROOT_PREFIX, properties, s3, s3Async));
+ storageCredentials.stream()
+ .filter(c -> c.prefix().startsWith(ROOT_PREFIX))
+ .collect(Collectors.toList())
+ .forEach(
+ storageCredential -> {
+ Map<String, String> propertiesWithCredentials =
+ ImmutableMap.<String, String>builder()
+ .putAll(properties)
+ .putAll(storageCredential.config())
+ .buildKeepingLast();
+
+ clientByPrefix.put(
+ storageCredential.prefix(),
+ new PrefixedS3Client(
+ storageCredential.prefix(), propertiesWithCredentials, s3, s3Async));
+ });
}
}
}
- return asyncClient;
- }
- private boolean shouldUseAsyncClient() {
- return s3FileIOProperties.isS3AnalyticsAcceleratorEnabled();
+ return clientByPrefix;
}
private ExecutorService executorService() {
@@ -423,7 +466,8 @@ public class S3FileIO
if (executorService == null) {
executorService =
ThreadPools.newExitingWorkerPool(
- "iceberg-s3fileio-delete", s3FileIOProperties.deleteThreads());
+ "iceberg-s3fileio-delete",
+ clientForStoragePath(ROOT_PREFIX).s3FileIOProperties().deleteThreads());
}
}
}
@@ -439,44 +483,17 @@ public class S3FileIO
@Override
public void initialize(Map<String, String> props) {
this.properties = SerializableMap.copyOf(props);
- Map<String, String> propertiesWithCredentials =
- ImmutableMap.<String, String>builder()
- .putAll(properties)
- .putAll(storageCredentialConfig())
- .buildKeepingLast();
- this.s3FileIOProperties = new S3FileIOProperties(propertiesWithCredentials);
this.createStack =
- PropertyUtil.propertyAsBoolean(props, "init-creation-stacktrace", true)
+ PropertyUtil.propertyAsBoolean(properties, "init-creation-stacktrace", true)
? Thread.currentThread().getStackTrace()
: null;
- // Do not override s3 client if it was provided
if (s3 == null) {
- Object clientFactory = S3FileIOAwsClientFactories.initialize(props);
- if (clientFactory instanceof S3FileIOAwsClientFactory) {
- this.s3 = ((S3FileIOAwsClientFactory) clientFactory)::s3;
- }
- if (clientFactory instanceof AwsClientFactory) {
- this.s3 = ((AwsClientFactory) clientFactory)::s3;
- }
+ Object clientFactory = S3FileIOAwsClientFactories.initialize(properties);
if (clientFactory instanceof CredentialSupplier) {
this.credential = ((CredentialSupplier) clientFactory).getCredential();
}
- if (s3FileIOProperties.isPreloadClientEnabled()) {
- client();
- }
- }
-
- // Do not override s3Async client if it was provided
- if (s3Async == null) {
- Object clientFactory = S3FileIOAwsClientFactories.initialize(props);
- if (clientFactory instanceof S3FileIOAwsClientFactory) {
- this.s3Async = ((S3FileIOAwsClientFactory) clientFactory)::s3Async;
- }
- if (clientFactory instanceof AwsClientFactory) {
- this.s3Async = ((AwsClientFactory) clientFactory)::s3Async;
- }
}
initMetrics(properties);
@@ -490,7 +507,7 @@ public class S3FileIO
DynConstructors.builder(MetricsContext.class)
.hiddenImpl(DEFAULT_METRICS_IMPL, String.class)
.buildChecked();
- MetricsContext context = ctor.newInstance("s3");
+ MetricsContext context = ctor.newInstance(ROOT_PREFIX);
context.initialize(props);
this.metrics = context;
} catch (NoClassDefFoundError | NoSuchMethodException | ClassCastException e) {
@@ -503,13 +520,9 @@ public class S3FileIO
public void close() {
// handles concurrent calls to close()
if (isResourceClosed.compareAndSet(false, true)) {
- if (client != null) {
- client.close();
- }
- if (asyncClient != null) {
- // cleanup usage in analytics accelerator if any
- AnalyticsAcceleratorUtil.cleanupCache(asyncClient, s3FileIOProperties);
- asyncClient.close();
+ if (clientByPrefix != null) {
+ clientByPrefix.values().forEach(PrefixedS3Client::close);
+ this.clientByPrefix = null;
}
}
}
@@ -531,9 +544,11 @@ public class S3FileIO
@Override
public boolean recoverFile(String path) {
- S3URI location = new S3URI(path, s3FileIOProperties.bucketToAccessPointMapping());
+ PrefixedS3Client client = clientForStoragePath(path);
+ S3URI location = new S3URI(path, client.s3FileIOProperties().bucketToAccessPointMapping());
ListObjectVersionsIterable response =
- client()
+ client
+ .s3()
.listObjectVersionsPaginator(
builder -> builder.bucket(location.bucket()).prefix(location.key()));
@@ -542,10 +557,12 @@ public class S3FileIO
Optional<ObjectVersion> recoverVersion =
response.versions().stream().max(Comparator.comparing(ObjectVersion::lastModified));
- return recoverVersion.map(version -> recoverObject(version, location.bucket())).orElse(false);
+ return recoverVersion
+ .map(version -> recoverObject(client, version, location.bucket()))
+ .orElse(false);
}
- private boolean recoverObject(ObjectVersion version, String bucket) {
+ private boolean recoverObject(PrefixedS3Client client, ObjectVersion version, String bucket) {
if (version.isLatest()) {
return true;
}
@@ -554,7 +571,8 @@ public class S3FileIO
try {
// Perform a copy instead of deleting the delete marker
// so that recovery does not rely on delete permissions
- client()
+ client
+ .s3()
.copyObject(
builder ->
builder
@@ -582,16 +600,4 @@ public class S3FileIO
public List<StorageCredential> credentials() {
return ImmutableList.copyOf(storageCredentials);
}
-
- private Map<String, String> storageCredentialConfig() {
- List<StorageCredential> s3Credentials =
- storageCredentials.stream()
- .filter(c -> c.prefix().startsWith("s3"))
- .collect(Collectors.toList());
-
- Preconditions.checkState(
- s3Credentials.size() <= 1, "Invalid S3 Credentials: only one S3 credential should exist");
-
- return s3Credentials.isEmpty() ? Map.of() : s3Credentials.get(0).config();
- }
}
diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIOProperties.java b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIOProperties.java
index 7f266f099d..928118346f 100644
--- a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIOProperties.java
+++ b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIOProperties.java
@@ -1174,4 +1174,8 @@ public class S3FileIOProperties implements Serializable {
e);
}
}
+
+ public Map<String, String> properties() {
+ return allProperties;
+ }
}
diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputFile.java b/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputFile.java
index e86bf4db39..63ce84e2fe 100644
--- a/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputFile.java
+++ b/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputFile.java
@@ -30,6 +30,13 @@ public class S3InputFile extends BaseS3File implements InputFile, NativelyEncryp
private NativeFileCryptoParameters nativeDecryptionParameters;
private Long length;
+ /**
+ * Creates a {@link S3InputFile} from the given parameters.
+ *
+ * @deprecated since 1.10.0, will be removed in 1.11.0; use {@link
+ * S3InputFile#fromLocation(String, PrefixedS3Client, MetricsContext)} instead.
+ */
+ @Deprecated
public static S3InputFile fromLocation(
String location,
S3Client client,
@@ -44,6 +51,13 @@ public class S3InputFile extends BaseS3File implements InputFile, NativelyEncryp
metrics);
}
+ /**
+ * Creates a {@link S3InputFile} from the given parameters.
+ *
+ * @deprecated since 1.10.0, will be removed in 1.11.0; use {@link
+ * S3InputFile#fromLocation(String, PrefixedS3Client, MetricsContext)} instead.
+ */
+ @Deprecated
public static S3InputFile fromLocation(
String location,
S3Client client,
@@ -59,6 +73,13 @@ public class S3InputFile extends BaseS3File implements InputFile, NativelyEncryp
metrics);
}
+ /**
+ * Creates a {@link S3InputFile} from the given parameters.
+ *
+ * @deprecated since 1.10.0, will be removed in 1.11.0; use {@link
+ * S3InputFile#fromLocation(String, long, PrefixedS3Client, MetricsContext)} instead.
+ */
+ @Deprecated
public static S3InputFile fromLocation(
String location,
long length,
@@ -74,6 +95,13 @@ public class S3InputFile extends BaseS3File implements InputFile, NativelyEncryp
metrics);
}
+ /**
+ * Creates a {@link S3InputFile} from the given parameters.
+ *
+ * @deprecated since 1.10.0, will be removed in 1.11.0; use {@link
+ * S3InputFile#fromLocation(String, long, PrefixedS3Client, MetricsContext)} instead.
+ */
+ @Deprecated
public static S3InputFile fromLocation(
String location,
long length,
@@ -90,6 +118,22 @@ public class S3InputFile extends BaseS3File implements InputFile, NativelyEncryp
metrics);
}
+ static S3InputFile fromLocation(
+ String location, PrefixedS3Client client, MetricsContext metrics) {
+ return fromLocation(location, 0, client, metrics);
+ }
+
+ static S3InputFile fromLocation(
+ String location, long length, PrefixedS3Client client, MetricsContext metrics) {
+ return new S3InputFile(
+ client.s3(),
+ client.s3FileIOProperties().isS3AnalyticsAcceleratorEnabled() ? client.s3Async() : null,
+ new S3URI(location, client.s3FileIOProperties().bucketToAccessPointMapping()),
+ length > 0 ? length : null,
+ client.s3FileIOProperties(),
+ metrics);
+ }
+
S3InputFile(
S3Client client,
S3AsyncClient asyncClient,
diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputFile.java b/aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputFile.java
index af69c1b868..fa12bac311 100644
--- a/aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputFile.java
+++ b/aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputFile.java
@@ -33,6 +33,13 @@ import software.amazon.awssdk.services.s3.S3Client;
public class S3OutputFile extends BaseS3File implements OutputFile, NativelyEncryptedFile {
private NativeFileCryptoParameters nativeEncryptionParameters;
+ /**
+ * Creates a {@link S3OutputFile} from the given parameters.
+ *
+ * @deprecated since 1.10.0, will be removed in 1.11.0; use {@link
+ * S3OutputFile#fromLocation(String, PrefixedS3Client, MetricsContext)} instead.
+ */
+ @Deprecated
public static S3OutputFile fromLocation(
String location,
S3Client client,
@@ -46,6 +53,13 @@ public class S3OutputFile extends BaseS3File implements OutputFile, NativelyEncr
metrics);
}
+ /**
+ * Creates a {@link S3OutputFile} from the given parameters.
+ *
+ * @deprecated since 1.10.0, will be removed in 1.11.0; use {@link
+ * S3OutputFile#fromLocation(String, PrefixedS3Client, MetricsContext)} instead.
+ */
+ @Deprecated
public static S3OutputFile fromLocation(
String location,
S3Client client,
@@ -60,6 +74,16 @@ public class S3OutputFile extends BaseS3File implements OutputFile, NativelyEncr
metrics);
}
+ static S3OutputFile fromLocation(
+ String location, PrefixedS3Client client, MetricsContext metrics) {
+ return new S3OutputFile(
+ client.s3(),
+ client.s3FileIOProperties().isS3AnalyticsAcceleratorEnabled() ? client.s3Async() : null,
+ new S3URI(location, client.s3FileIOProperties().bucketToAccessPointMapping()),
+ client.s3FileIOProperties(),
+ metrics);
+ }
+
S3OutputFile(
S3Client client,
S3AsyncClient asyncClient,
diff --git a/aws/src/test/java/org/apache/iceberg/aws/s3/TestPrefixedS3Client.java b/aws/src/test/java/org/apache/iceberg/aws/s3/TestPrefixedS3Client.java
new file mode 100644
index 0000000000..18be1c915b
--- /dev/null
+++ b/aws/src/test/java/org/apache/iceberg/aws/s3/TestPrefixedS3Client.java
@@ -0,0 +1,60 @@
+/*
+ * 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.iceberg.aws.s3;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.Map;
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Client;
+
+@SuppressWarnings("resource")
+public class TestPrefixedS3Client {
+
+ @Test
+ public void invalidParameters() {
+ assertThatThrownBy(() -> new PrefixedS3Client(null, null, null, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid storage prefix: null or empty");
+
+ assertThatThrownBy(() -> new PrefixedS3Client("", null, null, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid storage prefix: null or empty");
+
+ assertThatThrownBy(() -> new PrefixedS3Client("s3://bucket", null, null, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid properties: null");
+ }
+
+ @Test
+ public void validParameters() {
+ Map<String, String> properties =
+ ImmutableMap.of(AwsClientProperties.CLIENT_REGION, "us-east-1");
+ PrefixedS3Client client = new PrefixedS3Client("s3", properties, null, null);
+ assertThat(client.storagePrefix()).isEqualTo("s3");
+ assertThat(client.s3FileIOProperties().properties())
+ .isEqualTo(new S3FileIOProperties(properties).properties());
+ assertThat(client.s3()).isInstanceOf(S3Client.class);
+ assertThat(client.s3Async()).isInstanceOf(S3AsyncClient.class);
+ }
+}