diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java index b787cb8f4247..48a4f860e8bb 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java @@ -23,7 +23,8 @@ import java.util.UUID; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; -import org.apache.iceberg.aws.AwsClientUtil; +import org.apache.iceberg.aws.AwsClientFactories; +import org.apache.iceberg.aws.AwsClientFactory; import org.apache.iceberg.aws.AwsIntegTestUtil; import org.apache.iceberg.aws.AwsProperties; import org.apache.iceberg.aws.s3.S3FileIO; @@ -51,8 +52,9 @@ public class GlueTestBase { static final List namespaces = Lists.newArrayList(); // aws clients - static final GlueClient glue = AwsClientUtil.defaultGlueClient(); - static final S3Client s3 = AwsClientUtil.defaultS3Client(); + static final AwsClientFactory clientFactory = AwsClientFactories.defaultFactory(); + static final GlueClient glue = clientFactory.glue(); + static final S3Client s3 = clientFactory.s3(); // iceberg static GlueCatalog glueCatalog; @@ -64,13 +66,13 @@ public class GlueTestBase { @BeforeClass public static void beforeClass() { String testBucketPath = "s3://" + testBucketName + "/" + testPathPrefix; - S3FileIO fileIO = new S3FileIO(); - glueCatalog = new GlueCatalog(glue); - glueCatalog.initialize(catalogName, testBucketPath, new AwsProperties(), fileIO); + S3FileIO fileIO = new S3FileIO(clientFactory::s3); + glueCatalog = new GlueCatalog(); + glueCatalog.initialize(catalogName, testBucketPath, new AwsProperties(), glue, fileIO); AwsProperties properties = new AwsProperties(); properties.setGlueCatalogSkipArchive(true); - glueCatalogWithSkip = new GlueCatalog(glue); - glueCatalogWithSkip.initialize(catalogName, testBucketPath, properties, fileIO); + glueCatalogWithSkip = new GlueCatalog(); + glueCatalogWithSkip.initialize(catalogName, testBucketPath, properties, glue, fileIO); } @AfterClass diff --git a/aws/src/integration/java/org/apache/iceberg/aws/s3/S3FileIOTest.java b/aws/src/integration/java/org/apache/iceberg/aws/s3/S3FileIOTest.java index e97154cfb4e7..06e18b8426c9 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/s3/S3FileIOTest.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/s3/S3FileIOTest.java @@ -29,11 +29,14 @@ import java.util.UUID; import javax.crypto.KeyGenerator; import javax.crypto.SecretKey; -import org.apache.iceberg.aws.AwsClientUtil; +import org.apache.commons.lang3.SerializationUtils; +import org.apache.iceberg.aws.AwsClientFactories; +import org.apache.iceberg.aws.AwsClientFactory; import org.apache.iceberg.aws.AwsIntegTestUtil; import org.apache.iceberg.aws.AwsProperties; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.junit.AfterClass; import org.junit.Assert; import org.junit.Before; @@ -57,6 +60,7 @@ public class S3FileIOTest { + private static AwsClientFactory clientFactory; private static S3Client s3; private static KmsClient kms; private static String bucketName; @@ -69,8 +73,9 @@ public class S3FileIOTest { @BeforeClass public static void beforeClass() { - s3 = AwsClientUtil.defaultS3Client(); - kms = AwsClientUtil.defaultKmsClient(); + clientFactory = AwsClientFactories.defaultFactory(); + s3 = clientFactory.s3(); + kms = clientFactory.kms(); bucketName = AwsIntegTestUtil.testBucketName(); prefix = UUID.randomUUID().toString(); contentBytes = new byte[1024 * 1024 * 10]; @@ -94,13 +99,13 @@ public void before() { public void testNewInputStream() throws Exception { s3.putObject(PutObjectRequest.builder().bucket(bucketName).key(objectKey).build(), RequestBody.fromBytes(contentBytes)); - S3FileIO s3FileIO = new S3FileIO(); + S3FileIO s3FileIO = new S3FileIO(clientFactory::s3); validateRead(s3FileIO); } @Test public void testNewOutputStream() throws Exception { - S3FileIO s3FileIO = new S3FileIO(); + S3FileIO s3FileIO = new S3FileIO(clientFactory::s3); write(s3FileIO); InputStream stream = s3.getObject(GetObjectRequest.builder().bucket(bucketName).key(objectKey).build()); String result = IoUtils.toUtf8String(stream); @@ -112,7 +117,7 @@ public void testNewOutputStream() throws Exception { public void testSSE_S3() throws Exception { AwsProperties properties = new AwsProperties(); properties.setS3FileIoSseType(AwsProperties.S3FILEIO_SSE_TYPE_S3); - S3FileIO s3FileIO = new S3FileIO(AwsClientUtil::defaultS3Client, properties); + S3FileIO s3FileIO = new S3FileIO(clientFactory::s3, properties); write(s3FileIO); validateRead(s3FileIO); GetObjectResponse response = s3.getObject( @@ -125,7 +130,7 @@ public void testSSE_KMS() throws Exception { AwsProperties properties = new AwsProperties(); properties.setS3FileIoSseType(AwsProperties.S3FILEIO_SSE_TYPE_KMS); properties.setS3FileIoSseKey(kmsKeyArn); - S3FileIO s3FileIO = new S3FileIO(AwsClientUtil::defaultS3Client, properties); + S3FileIO s3FileIO = new S3FileIO(clientFactory::s3, properties); write(s3FileIO); validateRead(s3FileIO); GetObjectResponse response = s3.getObject( @@ -138,7 +143,7 @@ public void testSSE_KMS() throws Exception { public void testSSE_KMS_default() throws Exception { AwsProperties properties = new AwsProperties(); properties.setS3FileIoSseType(AwsProperties.S3FILEIO_SSE_TYPE_KMS); - S3FileIO s3FileIO = new S3FileIO(AwsClientUtil::defaultS3Client, properties); + S3FileIO s3FileIO = new S3FileIO(clientFactory::s3, properties); write(s3FileIO); validateRead(s3FileIO); GetObjectResponse response = s3.getObject( @@ -167,7 +172,7 @@ public void testSSE_Custom() throws Exception { properties.setS3FileIoSseType(AwsProperties.S3FILEIO_SSE_TYPE_CUSTOM); properties.setS3FileIoSseKey(encodedKey); properties.setS3FileIoSseMd5(md5); - S3FileIO s3FileIO = new S3FileIO(AwsClientUtil::defaultS3Client, properties); + S3FileIO s3FileIO = new S3FileIO(clientFactory::s3, properties); write(s3FileIO); validateRead(s3FileIO); GetObjectResponse response = s3.getObject( @@ -185,7 +190,7 @@ public void testSSE_Custom() throws Exception { public void testACL() throws Exception { AwsProperties properties = new AwsProperties(); properties.setS3FileIoAcl(ObjectCannedACL.BUCKET_OWNER_FULL_CONTROL); - S3FileIO s3FileIO = new S3FileIO(AwsClientUtil::defaultS3Client, properties); + S3FileIO s3FileIO = new S3FileIO(clientFactory::s3, properties); write(s3FileIO); validateRead(s3FileIO); GetObjectAclResponse response = s3.getObjectAcl( @@ -195,6 +200,16 @@ public void testACL() throws Exception { Assert.assertEquals(Permission.FULL_CONTROL, response.grants().get(0).permission()); } + @Test + public void testClientFactorySerialization() throws Exception { + S3FileIO fileIO = new S3FileIO(); + fileIO.initialize(Maps.newHashMap()); + write(fileIO); + byte [] data = SerializationUtils.serialize(fileIO); + S3FileIO fileIO2 = SerializationUtils.deserialize(data); + validateRead(fileIO2); + } + private void write(S3FileIO s3FileIO) throws Exception { OutputFile outputFile = s3FileIO.newOutputFile(objectUri); OutputStream outputStream = outputFile.create(); diff --git a/aws/src/integration/java/org/apache/iceberg/aws/s3/S3MultipartUploadTest.java b/aws/src/integration/java/org/apache/iceberg/aws/s3/S3MultipartUploadTest.java index 5f636aded59d..6054ec117d30 100644 --- a/aws/src/integration/java/org/apache/iceberg/aws/s3/S3MultipartUploadTest.java +++ b/aws/src/integration/java/org/apache/iceberg/aws/s3/S3MultipartUploadTest.java @@ -24,7 +24,7 @@ import java.util.UUID; import java.util.function.Supplier; import java.util.stream.IntStream; -import org.apache.iceberg.aws.AwsClientUtil; +import org.apache.iceberg.aws.AwsClientFactories; import org.apache.iceberg.aws.AwsIntegTestUtil; import org.apache.iceberg.aws.AwsProperties; import org.apache.iceberg.io.PositionOutputStream; @@ -51,7 +51,7 @@ public class S3MultipartUploadTest { @BeforeClass public static void beforeClass() { - s3 = AwsClientUtil.defaultS3Client(); + s3 = AwsClientFactories.defaultFactory().s3(); bucketName = AwsIntegTestUtil.testBucketName(); prefix = UUID.randomUUID().toString(); properties = new AwsProperties(); diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java b/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java new file mode 100644 index 000000000000..47205cf7696a --- /dev/null +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java @@ -0,0 +1,102 @@ +/* + * 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; + +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.FileIO; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.kms.KmsClient; +import software.amazon.awssdk.services.s3.S3Client; + +public class AwsClientFactories { + + /** + * The implementation class of {@link AwsClientFactory} to customize AWS client configurations. + * If set, all AWS clients will be configured by the specified class before initialization. + */ + public static final String CLIENT_FACTORY_CONFIG_KEY = "client.factory"; + + private static final SdkHttpClient HTTP_CLIENT_DEFAULT = UrlConnectionHttpClient.create(); + private static final DefaultAwsClientFactory AWS_CLIENT_FACTORY_DEFAULT = new DefaultAwsClientFactory(); + + private AwsClientFactories() { + } + + public static AwsClientFactory defaultFactory() { + return AWS_CLIENT_FACTORY_DEFAULT; + } + + public static AwsClientFactory from(Map properties) { + if (properties.containsKey(CLIENT_FACTORY_CONFIG_KEY)) { + return loadClientFactory(properties.get(CLIENT_FACTORY_CONFIG_KEY), properties); + } else { + return defaultFactory(); + } + } + + private static AwsClientFactory loadClientFactory(String impl, Map properties) { + DynConstructors.Ctor ctor; + try { + ctor = DynConstructors.builder(FileIO.class).impl(impl).buildChecked(); + } catch (NoSuchMethodException e) { + throw new IllegalArgumentException(String.format( + "Cannot initialize AwsClientFactory, missing no-arg constructor: %s", impl), e); + } + + AwsClientFactory factory; + try { + factory = ctor.newInstance(); + } catch (ClassCastException e) { + throw new IllegalArgumentException( + String.format("Cannot initialize AwsClientFactory, %s does not implement AwsClientFactory.", impl), e); + } + + factory.initialize(properties); + return factory; + } + + static class DefaultAwsClientFactory implements AwsClientFactory { + + DefaultAwsClientFactory() { + } + + @Override + public S3Client s3() { + return S3Client.builder().httpClient(HTTP_CLIENT_DEFAULT).build(); + } + + @Override + public GlueClient glue() { + return GlueClient.builder().httpClient(HTTP_CLIENT_DEFAULT).build(); + } + + @Override + public KmsClient kms() { + return KmsClient.builder().httpClient(HTTP_CLIENT_DEFAULT).build(); + } + + @Override + public void initialize(Map properties) { + } + } +} diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactory.java b/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactory.java new file mode 100644 index 000000000000..801559e81fa4 --- /dev/null +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsClientFactory.java @@ -0,0 +1,58 @@ +/* + * 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; + +import java.io.Serializable; +import java.util.Map; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.kms.KmsClient; +import software.amazon.awssdk.services.s3.S3Client; + +/** + * Interface to customize AWS clients used by Iceberg. + * A custom factory must have a no-arg constructor, and use {@link #initialize(Map)} to initialize the factory. + */ +public interface AwsClientFactory extends Serializable { + + /** + * create a Amazon S3 client + * @return s3 client + */ + S3Client s3(); + + /** + * create a AWS Glue client + * @return glue client + */ + GlueClient glue(); + + /** + * Create a AWS KMS client + * @return kms client + */ + KmsClient kms(); + + /** + * Initialize AWS client factory from catalog properties. + * @param properties catalog properties + */ + void initialize(Map properties); + +} diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsClientUtil.java b/aws/src/main/java/org/apache/iceberg/aws/AwsClientUtil.java deleted file mode 100644 index 71c7ed0db494..000000000000 --- a/aws/src/main/java/org/apache/iceberg/aws/AwsClientUtil.java +++ /dev/null @@ -1,57 +0,0 @@ -/* - * 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; - -import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient; -import software.amazon.awssdk.services.glue.GlueClient; -import software.amazon.awssdk.services.kms.KmsClient; -import software.amazon.awssdk.services.s3.S3Client; - -/** - * Provide factory methods to get default AWS clients. - * The clients use a default {@link UrlConnectionHttpClient} to avoid multiple versions of AWS HTTP client builders - * existing in the Java classpath, causing non-deterministic behavior of the AWS client. - * The credential and region information are both loaded from the default chain. - * For more details, see https://docs.aws.amazon.com/sdk-for-java/v2/developer-guide/credentials.html and - * https://docs.aws.amazon.com/sdk-for-java/v2/developer-guide/java-dg-region-selection.html - */ -public class AwsClientUtil { - - private AwsClientUtil() { - } - - public static S3Client defaultS3Client() { - return S3Client.builder() - .httpClient(UrlConnectionHttpClient.create()) - .build(); - } - - public static KmsClient defaultKmsClient() { - return KmsClient.builder() - .httpClient(UrlConnectionHttpClient.create()) - .build(); - } - - public static GlueClient defaultGlueClient() { - return GlueClient.builder() - .httpClient(UrlConnectionHttpClient.create()) - .build(); - } -} diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java b/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java index d7ac5a023de8..865471a9521b 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java @@ -19,12 +19,13 @@ package org.apache.iceberg.aws; +import java.io.Serializable; import java.util.Map; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.util.PropertyUtil; import software.amazon.awssdk.services.s3.model.ObjectCannedACL; -public class AwsProperties { +public class AwsProperties implements Serializable { /** * Type of S3 Server side encryption used, default to {@link AwsProperties#S3FILEIO_SSE_TYPE_NONE}. diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java index 86ce858a825d..2dddda9c6878 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java +++ b/aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java @@ -32,7 +32,7 @@ import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableOperations; -import org.apache.iceberg.aws.AwsClientUtil; +import org.apache.iceberg.aws.AwsClientFactories; import org.apache.iceberg.aws.AwsProperties; import org.apache.iceberg.aws.s3.S3FileIO; import org.apache.iceberg.catalog.Namespace; @@ -72,7 +72,7 @@ public class GlueCatalog extends BaseMetastoreCatalog implements Closeable, Supp private static final Logger LOG = LoggerFactory.getLogger(GlueCatalog.class); - private final GlueClient glue; + private GlueClient glue; private Configuration hadoopConf; private String catalogName; private String warehousePath; @@ -82,33 +82,38 @@ public class GlueCatalog extends BaseMetastoreCatalog implements Closeable, Supp /** * No-arg constructor to load the catalog dynamically. *

- * Only the AWS Glue client is initialized. - * Other fields must be initialized by calling {@link GlueCatalog#initialize(String, Map)} later. + * All fields are initialized by calling {@link GlueCatalog#initialize(String, Map)} later. */ public GlueCatalog() { - this(AwsClientUtil.defaultGlueClient()); - } - - @VisibleForTesting - GlueCatalog(GlueClient glue) { - this.glue = glue; } @Override public void initialize(String name, Map properties) { - String fileIOImpl = properties.get(CatalogProperties.FILE_IO_IMPL); initialize( name, properties.get(CatalogProperties.WAREHOUSE_LOCATION), new AwsProperties(properties), - fileIOImpl == null ? new S3FileIO() : CatalogUtil.loadFileIO(fileIOImpl, properties, hadoopConf)); + AwsClientFactories.from(properties).glue(), + initializeFileIO(properties)); + } + + private FileIO initializeFileIO(Map properties) { + String fileIOImpl = properties.get(CatalogProperties.FILE_IO_IMPL); + if (fileIOImpl == null) { + FileIO io = new S3FileIO(); + io.initialize(properties); + return io; + } else { + return CatalogUtil.loadFileIO(fileIOImpl, properties, hadoopConf); + } } @VisibleForTesting - void initialize(String name, String path, AwsProperties properties, FileIO io) { + void initialize(String name, String path, AwsProperties properties, GlueClient client, FileIO io) { this.catalogName = name; this.awsProperties = properties; this.warehousePath = cleanWarehousePath(path); + this.glue = client; this.fileIO = io; } 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 469f93532b82..25a1b1b7d843 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 @@ -20,7 +20,8 @@ package org.apache.iceberg.aws.s3; import java.util.Map; -import org.apache.iceberg.aws.AwsClientUtil; +import org.apache.iceberg.aws.AwsClientFactories; +import org.apache.iceberg.aws.AwsClientFactory; import org.apache.iceberg.aws.AwsProperties; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.InputFile; @@ -39,18 +40,36 @@ * Using this FileIO with other schemes will result in {@link org.apache.iceberg.exceptions.ValidationException}. */ public class S3FileIO implements FileIO { - private final SerializableSupplier s3; + private SerializableSupplier s3; private AwsProperties awsProperties; + private AwsClientFactory awsClientFactory; private transient S3Client client; + /** + * No-arg constructor to load the FileIO dynamically. + *

+ * All fields are initialized by calling {@link S3FileIO#initialize(Map)} later. + */ public S3FileIO() { - this(AwsClientUtil::defaultS3Client); } + /** + * Constructor with custom s3 supplier and default AWS properties. + *

+ * Calling {@link S3FileIO#initialize(Map)} will overwrite information set in this constructor. + * @param s3 s3 supplier + */ public S3FileIO(SerializableSupplier s3) { this(s3, new AwsProperties()); } + /** + * Constructor with custom s3 supplier and AWS properties. + *

+ * Calling {@link S3FileIO#initialize(Map)} will overwrite information set in this constructor. + * @param s3 s3 supplier + * @param awsProperties aws properties + */ public S3FileIO(SerializableSupplier s3, AwsProperties awsProperties) { this.s3 = s3; this.awsProperties = awsProperties; @@ -87,5 +106,7 @@ private S3Client client() { @Override public void initialize(Map properties) { this.awsProperties = new AwsProperties(properties); + this.awsClientFactory = AwsClientFactories.from(properties); + this.s3 = awsClientFactory::s3; } } diff --git a/aws/src/test/java/org/apache/iceberg/aws/AwsClientFactoriesTest.java b/aws/src/test/java/org/apache/iceberg/aws/AwsClientFactoriesTest.java new file mode 100644 index 000000000000..c0b48cd8e223 --- /dev/null +++ b/aws/src/test/java/org/apache/iceberg/aws/AwsClientFactoriesTest.java @@ -0,0 +1,76 @@ +/* + * 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; + +import java.util.Map; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.junit.Assert; +import org.junit.Test; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.kms.KmsClient; +import software.amazon.awssdk.services.s3.S3Client; + + +public class AwsClientFactoriesTest { + + @Test + public void testLoadDefault() { + Assert.assertEquals("default client should be singleton", + AwsClientFactories.defaultFactory(), AwsClientFactories.defaultFactory()); + + Assert.assertTrue("should load default when not configured", + AwsClientFactories.from(Maps.newHashMap()) instanceof AwsClientFactories.DefaultAwsClientFactory); + } + + @Test + public void testLoadCustom() { + Map properties = Maps.newHashMap(); + properties.put(AwsClientFactories.CLIENT_FACTORY_CONFIG_KEY, CustomFactory.class.getName()); + Assert.assertTrue("should load custom class", + AwsClientFactories.from(properties) instanceof CustomFactory); + } + + public static class CustomFactory implements AwsClientFactory { + + public CustomFactory() { + } + + @Override + public S3Client s3() { + return null; + } + + @Override + public GlueClient glue() { + return null; + } + + @Override + public KmsClient kms() { + return null; + } + + @Override + public void initialize(Map properties) { + + } + } + +} diff --git a/aws/src/test/java/org/apache/iceberg/aws/glue/GlueCatalogTest.java b/aws/src/test/java/org/apache/iceberg/aws/glue/GlueCatalogTest.java index a80b168f77bb..45329c1c1e77 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/glue/GlueCatalogTest.java +++ b/aws/src/test/java/org/apache/iceberg/aws/glue/GlueCatalogTest.java @@ -69,8 +69,8 @@ public class GlueCatalogTest { @Before public void before() { glue = Mockito.mock(GlueClient.class); - glueCatalog = new GlueCatalog(glue); - glueCatalog.initialize(CATALOG_NAME, WAREHOUSE_PATH, new AwsProperties(), null); + glueCatalog = new GlueCatalog(); + glueCatalog.initialize(CATALOG_NAME, WAREHOUSE_PATH, new AwsProperties(), glue, null); } @Test @@ -79,15 +79,15 @@ public void constructor_emptyWarehousePath() { IllegalArgumentException.class, "Cannot initialize GlueCatalog because warehousePath must not be null", () -> { - GlueCatalog catalog = new GlueCatalog(glue); - catalog.initialize(CATALOG_NAME, null, new AwsProperties(), null); + GlueCatalog catalog = new GlueCatalog(); + catalog.initialize(CATALOG_NAME, null, new AwsProperties(), glue, null); }); } @Test public void constructor_warehousePathWithEndSlash() { - GlueCatalog catalogWithSlash = new GlueCatalog(glue); - catalogWithSlash.initialize(CATALOG_NAME, WAREHOUSE_PATH + "/", new AwsProperties(), null); + GlueCatalog catalogWithSlash = new GlueCatalog(); + catalogWithSlash.initialize(CATALOG_NAME, WAREHOUSE_PATH + "/", new AwsProperties(), glue, null); Mockito.doReturn(GetDatabaseResponse.builder() .database(Database.builder().name("db").build()).build()) .when(glue).getDatabase(Mockito.any(GetDatabaseRequest.class));