Distributed under the OSI-approved BSD 3-Clause License. See * http://ncip.github.com/HPC/LICENSE.txt for details. */ -package gov.nih.nci.hpc.integration.s3.impl; +package gov.nih.nci.hpc.integration.s3.v1.impl; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; diff --git a/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3Connection.java b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3Connection.java index 7eefae16c7..20d01e0bc7 100644 --- a/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3Connection.java +++ b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3Connection.java @@ -31,30 +31,30 @@ import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.core.exception.SdkException; -import software.amazon.awssdk.crt.CrtRuntimeException; -import software.amazon.awssdk.crt.Log; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.S3Configuration; -import software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder; import software.amazon.awssdk.services.s3.presigner.S3Presigner; import software.amazon.awssdk.transfer.s3.S3TransferManager; /** - * HPC S3 Connection. + * HPC S3 Connection base class. Holds all logic shared between the concrete S3 + * connection implementations. The only behavior that differs between + * implementations is how the {@link S3AsyncClient} is built (e.g. AWS CRT vs + * Netty-NIO), which is delegated to the abstract build methods. * * @author Eran Rosenberg */ -public class HpcS3Connection { +public abstract class HpcS3Connection { // ---------------------------------------------------------------------// // Constants // ---------------------------------------------------------------------// // 5GB in bytes - private static final long FIVE_GB = 5368709120L; + protected static final long FIVE_GB = 5368709120L; // Google Storage S3 URL. - private static final String GOOGLE_STORAGE_URL = "https://storage.googleapis.com"; + protected static final String GOOGLE_STORAGE_URL = "https://storage.googleapis.com"; // ---------------------------------------------------------------------// // Instance members @@ -66,25 +66,33 @@ public class HpcS3Connection { // The multipart upload minimum part size. @Value("${hpc.integration.s3.minimumUploadPartSize}") - private Long minimumUploadPartSize = null; + protected Long minimumUploadPartSize = null; // The multipart upload threshold. @Value("${hpc.integration.s3.multipartUploadThreshold}") - private Long multipartUploadThreshold = null; + protected Long multipartUploadThreshold = null; - // The CRT log file (Optional). - @Value("${hpc.integration.s3.crtLogFile:#{null}}") - private String crtLogFile = null; + // The connection timeout - the time to wait to establish a new connection, in + // milliseconds. Same behavior across all S3 client implementations, so the + // property is shared. + @Value("${hpc.integration.s3.connectionTimeout}") + protected Integer connectionTimeout = null; - // AWS CRT to trust all certs config. - @Value("${hpc.integration.s3.trustAllCerts:false}") - private Boolean trustAllCerts = null; + // The max number of error retries. Same behavior across all S3 client + // implementations, so the property is shared. + @Value("${hpc.integration.s3.maxErrorRetries}") + protected Integer maxErrorRetries = null; + + // Disable SSL certificate checking (for development/testing only). Shares the + // property with the v1 client since the behavior is the same. + @Value("${hpc.integration.s3.disableCertChecking:false}") + protected Boolean disableCertChecking = null; // The executor service to be used by AWSTransferManager private ExecutorService executorService = null; // The logger instance. - private final Logger logger = LoggerFactory.getLogger(getClass().getName()); + protected final Logger logger = LoggerFactory.getLogger(getClass().getName()); // ---------------------------------------------------------------------// // Constructors @@ -99,7 +107,7 @@ public class HpcS3Connection { * @param awsTransferManagerThreadPoolSize The thread pool size to be used for * AWS transfer manager */ - private HpcS3Connection(String pathStyleAccessEnabledProviders, int awsTransferManagerThreadPoolSize) { + protected HpcS3Connection(String pathStyleAccessEnabledProviders, int awsTransferManagerThreadPoolSize) { for (String s3Provider : pathStyleAccessEnabledProviders.split(",")) { this.pathStyleAccessEnabledProviders.add(HpcIntegratedSystem.fromValue(s3Provider)); } @@ -110,7 +118,33 @@ private HpcS3Connection(String pathStyleAccessEnabledProviders, int awsTransferM } // ---------------------------------------------------------------------// - // Methods + // Abstract Methods + // ---------------------------------------------------------------------// + + /** + * Build a S3 async client for a 'S3 3rd Party Provider' account. + * + * @param credentialsProvider The S3 credentials provider. + * @param endpoint The S3 provider endpoint. + * @param pathStyleAccessEnabled true if the S3 3rd Party provider supports path + * style access. + * @param thresholdInBytes The multipart upload threshold in bytes. + * @return A S3 async client. + */ + protected abstract S3AsyncClient buildS3ProviderAsyncClient(StaticCredentialsProvider credentialsProvider, + URI endpoint, boolean pathStyleAccessEnabled, long thresholdInBytes); + + /** + * Build a S3 async client for an AWS S3 account. + * + * @param credentialsProvider The AWS credentials provider. + * @param region The AWS account region. + * @return A S3 async client. + */ + protected abstract S3AsyncClient buildAwsAsyncClient(StaticCredentialsProvider credentialsProvider, String region); + + // ---------------------------------------------------------------------// + // Public Methods // ---------------------------------------------------------------------// /** @@ -224,7 +258,7 @@ public HpcIntegratedSystem getS3Provider(Object authenticatedToken) throws HpcEx // Helper Methods // ---------------------------------------------------------------------// - private class HpcS3 { + protected class HpcS3 { private S3TransferManager transferManager = null; private S3AsyncClient client = null; private S3Presigner presigner = null; @@ -262,24 +296,12 @@ private Object authenticateS3Provider(String username, String password, String u HpcS3 s3 = new HpcS3(); s3.provider = s3Provider; - try { - // If configured, start the AWS CRT logger. - if (!StringUtils.isEmpty(crtLogFile)) { - Log.initLoggingToFile(Log.LogLevel.Trace, crtLogFile); - } + long thresholdInBytes = url.equalsIgnoreCase(GOOGLE_STORAGE_URL) ? FIVE_GB : multipartUploadThreshold; - // Instantiate a S3 async client. - S3CrtAsyncClientBuilder crtAsyncClientBuilder = S3AsyncClient.crtBuilder() - .credentialsProvider(s3ProviderCredentialsProvider).forcePathStyle(pathStyleAccessEnabled) - .endpointOverride(uri).minimumPartSizeInBytes(minimumUploadPartSize) - .thresholdInBytes(url.equalsIgnoreCase(GOOGLE_STORAGE_URL) ? FIVE_GB : multipartUploadThreshold).maxConcurrency(1); - - if (trustAllCerts) { - crtAsyncClientBuilder.httpConfiguration(builder -> builder.trustAllCertificatesEnabled(true)); - logger.warn("hpc.integration.s3.trustAllCerts property is set to true. CRT cert validation is off"); - } - - s3.client = crtAsyncClientBuilder.build(); + try { + // Instantiate a S3 async client (implementation specific). + s3.client = buildS3ProviderAsyncClient(s3ProviderCredentialsProvider, uri, pathStyleAccessEnabled, + thresholdInBytes); // Instantiate the S3 transfer manager. s3.transferManager = S3TransferManager.builder().s3Client(s3.client).executor(executorService).build(); @@ -292,7 +314,7 @@ private Object authenticateS3Provider(String username, String password, String u return s3; - } catch (SdkException | CrtRuntimeException e) { + } catch (SdkException e) { throw new HpcException( "[S3] Failed to authenticate S3 Provider: " + s3Provider.value() + "] - " + e.getMessage(), HpcErrorType.DATA_TRANSFER_ERROR, e); @@ -317,15 +339,8 @@ private Object authenticateAWS(String accessKey, String secretKey, String region s3.provider = HpcIntegratedSystem.AWS; try { - // If configured, start the AWS CRT logger. - if (!StringUtils.isEmpty(crtLogFile)) { - Log.initLoggingToFile(Log.LogLevel.Trace, crtLogFile); - } - - // Instantiate a S3 async client. - s3.client = S3AsyncClient.crtBuilder().credentialsProvider(awsCredentialsProvider).region(Region.of(region)) - .minimumPartSizeInBytes(minimumUploadPartSize) - .thresholdInBytes(multipartUploadThreshold).maxConcurrency(1).build(); + // Instantiate a S3 async client (implementation specific). + s3.client = buildAwsAsyncClient(awsCredentialsProvider, region); // Instantiate the S3 transfer manager. s3.transferManager = S3TransferManager.builder().s3Client(s3.client).executor(executorService).build(); @@ -336,7 +351,7 @@ private Object authenticateAWS(String accessKey, String secretKey, String region return s3; - } catch (SdkException | CrtRuntimeException e) { + } catch (SdkException e) { throw new HpcException("[S3] Failed to authenticate S3 in region " + region + "] - " + e.getMessage(), HpcErrorType.DATA_TRANSFER_ERROR, e); } diff --git a/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionCrtAsyncClient.java b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionCrtAsyncClient.java new file mode 100644 index 0000000000..282f28fd10 --- /dev/null +++ b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionCrtAsyncClient.java @@ -0,0 +1,164 @@ +/** + * HpcS3ConnectionCrtAsyncClient.java + * + *
+ * Copyright SVG, Inc. Copyright Leidos Biomedical Research, Inc + * + *
+ * Distributed under the OSI-approved BSD 3-Clause License. See + * http://ncip.github.com/HPC/LICENSE.txt for details. + */ +package gov.nih.nci.hpc.integration.s3.v2.impl; + +import java.net.URI; +import java.time.Duration; +import java.util.Arrays; + +import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.factory.annotation.Value; + +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.crt.CrtRuntimeException; +import software.amazon.awssdk.crt.Log; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3AsyncClient; +import software.amazon.awssdk.services.s3.crt.S3CrtHttpConfiguration; +import software.amazon.awssdk.services.s3.crt.S3CrtRetryConfiguration; + +/** + * HPC S3 Connection - AWS CRT based implementation. + * + * @author Eran Rosenberg + */ +public class HpcS3ConnectionCrtAsyncClient extends HpcS3Connection { + // ---------------------------------------------------------------------// + // Constants + // ---------------------------------------------------------------------// + + // The CRT log level to fall back to if the configured value is invalid. + private static final Log.LogLevel DEFAULT_CRT_LOG_LEVEL = Log.LogLevel.Info; + + // ---------------------------------------------------------------------// + // Instance members + // ---------------------------------------------------------------------// + + // The CRT log file (Optional). + @Value("${hpc.integration.s3.crtLogFile:#{null}}") + private String crtLogFile = null; + + // The CRT log level (used when crtLogFile is configured). + @Value("${hpc.integration.s3.crtLogLevel:Info}") + private String crtLogLevel = null; + + // ---------------------------------------------------------------------// + // Constructors + // ---------------------------------------------------------------------// + + /** + * Constructor for Spring Dependency Injection. + * + * @param pathStyleAccessEnabledProviders A list of S3 3rd party providers that + * require connection w/ path-style + * enabled. + * @param awsTransferManagerThreadPoolSize The thread pool size to be used for + * AWS transfer manager + */ + HpcS3ConnectionCrtAsyncClient(String pathStyleAccessEnabledProviders, int awsTransferManagerThreadPoolSize) { + super(pathStyleAccessEnabledProviders, awsTransferManagerThreadPoolSize); + } + + // ---------------------------------------------------------------------// + // HpcS3Connection Abstract Class Implementation + // ---------------------------------------------------------------------// + + @Override + protected S3AsyncClient buildS3ProviderAsyncClient(StaticCredentialsProvider credentialsProvider, URI endpoint, + boolean pathStyleAccessEnabled, long thresholdInBytes) { + try { + // If configured, start the AWS CRT logger. + initCrtLogging(); + + // Instantiate a S3 async client. + return S3AsyncClient.crtBuilder().credentialsProvider(credentialsProvider) + .forcePathStyle(pathStyleAccessEnabled).endpointOverride(endpoint) + .minimumPartSizeInBytes(minimumUploadPartSize).thresholdInBytes(thresholdInBytes) + .httpConfiguration(httpConfiguration(true)).retryConfiguration(retryConfiguration()).build(); + + } catch (CrtRuntimeException e) { + throw SdkException.create(e.getMessage(), e); + } + } + + @Override + protected S3AsyncClient buildAwsAsyncClient(StaticCredentialsProvider credentialsProvider, String region) { + try { + // If configured, start the AWS CRT logger. + initCrtLogging(); + + // Instantiate a S3 async client. + return S3AsyncClient.crtBuilder().credentialsProvider(credentialsProvider).region(Region.of(region)) + .minimumPartSizeInBytes(minimumUploadPartSize).thresholdInBytes(multipartUploadThreshold) + .httpConfiguration(httpConfiguration(false)).retryConfiguration(retryConfiguration()).build(); + + } catch (CrtRuntimeException e) { + throw SdkException.create(e.getMessage(), e); + } + } + + // ---------------------------------------------------------------------// + // Helper Methods + // ---------------------------------------------------------------------// + + /** + * Build the HTTP configuration to be applied to a CRT S3 async client. Applies + * the shared connection timeout and, when requested, the trust-all-certs + * setting. + * + * @param applyDisableCertChecking true to honor the disableCertChecking config. + * @return A CRT HTTP configuration. + */ + private S3CrtHttpConfiguration httpConfiguration(boolean applyDisableCertChecking) { + S3CrtHttpConfiguration.Builder builder = S3CrtHttpConfiguration.builder() + .connectionTimeout(Duration.ofMillis(connectionTimeout)); + + if (applyDisableCertChecking && Boolean.TRUE.equals(disableCertChecking)) { + builder.trustAllCertificatesEnabled(true); + logger.warn("hpc.integration.s3.disableCertChecking property is set to true. CRT cert validation is off"); + } + + return builder.build(); + } + + /** + * Build the retry configuration to be applied to a CRT S3 async client, honoring + * the shared max error retries config. + * + * @return A CRT retry configuration. + */ + private S3CrtRetryConfiguration retryConfiguration() { + return S3CrtRetryConfiguration.builder().numRetries(maxErrorRetries).build(); + } + + /** + * If configured, start the AWS CRT logger. An invalid log level is logged and + * defaulted rather than raised, so that a bad logging configuration doesn't fail + * the S3 authentication. + */ + private void initCrtLogging() { + if (StringUtils.isEmpty(crtLogFile)) { + return; + } + + Log.LogLevel logLevel = DEFAULT_CRT_LOG_LEVEL; + try { + logLevel = Log.LogLevel.valueOf(crtLogLevel); + + } catch (IllegalArgumentException e) { + logger.warn("Invalid hpc.integration.s3.crtLogLevel [{}]. Valid values are {}. Defaulting to [{}]", + crtLogLevel, Arrays.toString(Log.LogLevel.values()), DEFAULT_CRT_LOG_LEVEL); + } + + Log.initLoggingToFile(logLevel, crtLogFile); + } +} diff --git a/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionFactory.java b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionFactory.java new file mode 100644 index 0000000000..74202eeb52 --- /dev/null +++ b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionFactory.java @@ -0,0 +1,86 @@ +/** + * HpcS3ConnectionFactory.java + * + *
+ * Copyright SVG, Inc. Copyright Leidos Biomedical Research, Inc + * + *
+ * Distributed under the OSI-approved BSD 3-Clause License. See + * http://ncip.github.com/HPC/LICENSE.txt for details. + */ +package gov.nih.nci.hpc.integration.s3.v2.impl; + +import gov.nih.nci.hpc.domain.error.HpcErrorType; +import gov.nih.nci.hpc.exception.HpcException; + +/** + * Factory that instantiates the configured {@link HpcS3Connection} + * implementation. The concrete implementation (AWS CRT or Netty-NIO based) is + * selected via the {@code hpc.integration.s3.asyncClient} property. + * + * @author Eran Rosenberg + */ +public class HpcS3ConnectionFactory { + // ---------------------------------------------------------------------// + // Enum Types + // ---------------------------------------------------------------------// + + /** + * The supported S3 async client implementations. + */ + public enum HpcS3AsyncClient { + CRT, NETTYNIO + } + + // ---------------------------------------------------------------------// + // Constructors + // ---------------------------------------------------------------------// + + /** + * Default constructor disabled - this is a static factory. + */ + private HpcS3ConnectionFactory() { + } + + // ---------------------------------------------------------------------// + // Factory Methods + // ---------------------------------------------------------------------// + + /** + * Create a {@link HpcS3Connection} of the configured implementation. Invoked by + * Spring as a factory-method; the returned instance remains a Spring-managed + * bean, so its {@code @Value} fields are injected and its lifecycle callbacks + * are honored. + * + * @param asyncClient The S3 async client implementation to + * use (CRT or NETTYNIO, + * case-insensitive). + * @param pathStyleAccessEnabledProviders A list of S3 3rd party providers that + * require connection w/ path-style + * enabled. + * @param awsTransferManagerThreadPoolSize The thread pool size to be used for + * AWS transfer manager. + * @return A concrete {@link HpcS3Connection} instance. + * @throws HpcException If the configured async client value is invalid. + */ + public static HpcS3Connection create(String asyncClient, String pathStyleAccessEnabledProviders, + int awsTransferManagerThreadPoolSize) throws HpcException { + HpcS3AsyncClient implementation; + try { + implementation = HpcS3AsyncClient.valueOf(asyncClient.trim().toUpperCase()); + + } catch (IllegalArgumentException | NullPointerException e) { + throw new HpcException("Invalid hpc.integration.s3.asyncClient [" + asyncClient + + "]. Valid values are CRT, NETTYNIO", HpcErrorType.INVALID_REQUEST_INPUT, e); + } + + switch (implementation) { + case NETTYNIO: + return new HpcS3ConnectionNettyNioAsyncClient(pathStyleAccessEnabledProviders, + awsTransferManagerThreadPoolSize); + case CRT: + default: + return new HpcS3ConnectionCrtAsyncClient(pathStyleAccessEnabledProviders, awsTransferManagerThreadPoolSize); + } + } +} diff --git a/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionNettyNioAsyncClient.java b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionNettyNioAsyncClient.java new file mode 100644 index 0000000000..928934ab3a --- /dev/null +++ b/src/hpc-server/hpc-integration-impl/src/main/java/gov/nih/nci/hpc/integration/s3/v2/impl/HpcS3ConnectionNettyNioAsyncClient.java @@ -0,0 +1,221 @@ +/** + * HpcS3ConnectionNettyNioAsyncClient.java + * + *
+ * Copyright SVG, Inc. Copyright Leidos Biomedical Research, Inc + * + *
+ * Distributed under the OSI-approved BSD 3-Clause License. See
+ * http://ncip.github.com/HPC/LICENSE.txt for details.
+ */
+package gov.nih.nci.hpc.integration.s3.v2.impl;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.function.Consumer;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Value;
+
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.retry.RetryPolicy;
+import software.amazon.awssdk.http.SdkHttpConfigurationOption;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration;
+import software.amazon.awssdk.utils.AttributeMap;
+
+/**
+ * HPC S3 Connection - Netty-NIO based implementation.
+ *
+ * @author Eran Rosenberg
+ */
+public class HpcS3ConnectionNettyNioAsyncClient extends HpcS3Connection implements DisposableBean {
+ // ---------------------------------------------------------------------//
+ // Instance members
+ // ---------------------------------------------------------------------//
+
+ // The maximum number of concurrent S3 requests, i.e. the size of the HTTP
+ // connection pool. The AWS SDK default is 50.
+ @Value("${hpc.integration.s3.maxConnections}")
+ private Integer maxConnections = null;
+
+ // The maximum number of multipart parts in flight per transfer. Bounded below the
+ // AWS SDK default of 50, which on its own can consume the entire HTTP connection
+ // pool and starve every other transfer sharing it. Keep it at or below
+ // maxConnections divided by the number of concurrent transfers
+ // (hpc.integration.s3.executorThreadPoolSize), leaving headroom for the requests
+ // that are not part uploads/downloads.
+ @Value("${hpc.integration.s3.nettynio.maxInFlightParts:8}")
+ private Integer maxInFlightParts = null;
+
+ // The time to wait for a connection from the pool before failing the request, in
+ // milliseconds. Raised from the AWS SDK default of 10 seconds, which a request
+ // queued behind an in flight part is likely to exceed - waiting for the pool is
+ // preferable to failing the transfer.
+ @Value("${hpc.integration.s3.nettynio.connectionAcquisitionTimeout:60000}")
+ private Long connectionAcquisitionTimeout = null;
+
+ // The socket timeout - the max time to wait for data to be transferred over an
+ // established, open connection, in milliseconds. Maps to the Netty client's read
+ // and write timeouts. Shares the property with the v1 client since the behavior
+ // is the same.
+ @Value("${hpc.integration.s3.socketTimeout}")
+ private Integer socketTimeout = null;
+
+ // The TCP keep alive setting. Shares the property with the v1 client since the
+ // behavior is the same.
+ @Value("${hpc.integration.s3.useTcpKeepAlive}")
+ private Boolean useTcpKeepAlive = false;
+
+ // A single Netty-NIO HTTP client, shared by all the S3 async clients created by
+ // this connection. Netty allocates an event loop group and a connection pool per
+ // HTTP client instance, so a client per S3 connection would accumulate threads.
+ // Note that the AWS SDK treats an explicitly provided HTTP client as
+ // 'non-managed', i.e. it does not close it when the S3 client is closed, so this
+ // client remains usable for the lifetime of this connection and is closed on
+ // bean destruction.
+ private volatile SdkAsyncHttpClient httpClient = null;
+
+ // The lock guarding the lazy instantiation of the shared HTTP client. The
+ // client can't be instantiated by the constructor because it is configured w/
+ // property values that are injected after construction.
+ private final Object httpClientLock = new Object();
+
+ // ---------------------------------------------------------------------//
+ // Constructors
+ // ---------------------------------------------------------------------//
+
+ /**
+ * Constructor for Spring Dependency Injection.
+ *
+ * @param pathStyleAccessEnabledProviders A list of S3 3rd party providers that
+ * require connection w/ path-style
+ * enabled.
+ * @param awsTransferManagerThreadPoolSize The thread pool size to be used for
+ * AWS transfer manager
+ */
+ HpcS3ConnectionNettyNioAsyncClient(String pathStyleAccessEnabledProviders, int awsTransferManagerThreadPoolSize) {
+ super(pathStyleAccessEnabledProviders, awsTransferManagerThreadPoolSize);
+ }
+
+ // ---------------------------------------------------------------------//
+ // HpcS3Connection Abstract Class Implementation
+ // ---------------------------------------------------------------------//
+
+ @Override
+ protected S3AsyncClient buildS3ProviderAsyncClient(StaticCredentialsProvider credentialsProvider, URI endpoint,
+ boolean pathStyleAccessEnabled, long thresholdInBytes) {
+ // Instantiate a S3 async client (Netty-NIO based) w/ multipart enabled.
+ return S3AsyncClient.builder().credentialsProvider(credentialsProvider).forcePathStyle(pathStyleAccessEnabled)
+ .endpointOverride(endpoint).multipartEnabled(true)
+ .multipartConfiguration(multipartConfiguration(thresholdInBytes))
+ .overrideConfiguration(overrideConfiguration -> overrideConfiguration.retryPolicy(retryPolicy()))
+ .httpClient(getNettyNioHttpClient()).build();
+ }
+
+ @Override
+ protected S3AsyncClient buildAwsAsyncClient(StaticCredentialsProvider credentialsProvider, String region) {
+ // Instantiate a S3 async client (Netty-NIO based) w/ multipart enabled.
+ return S3AsyncClient.builder().credentialsProvider(credentialsProvider).region(Region.of(region))
+ .multipartEnabled(true).multipartConfiguration(multipartConfiguration(multipartUploadThreshold))
+ .overrideConfiguration(overrideConfiguration -> overrideConfiguration.retryPolicy(retryPolicy()))
+ .httpClient(getNettyNioHttpClient()).build();
+ }
+
+ // ---------------------------------------------------------------------//
+ // DisposableBean Implementation
+ // ---------------------------------------------------------------------//
+
+ @Override
+ public void destroy() {
+ synchronized (httpClientLock) {
+ if (httpClient != null) {
+ logger.info("Closing the shared Netty-NIO HTTP client");
+ httpClient.close();
+ httpClient = null;
+ }
+ }
+ }
+
+ // ---------------------------------------------------------------------//
+ // Helper Methods
+ // ---------------------------------------------------------------------//
+
+ /**
+ * Get the multipart configuration to be applied to a S3 async client.
+ *
+ * @param thresholdInBytes The multipart upload threshold in bytes.
+ * @return A multipart configuration consumer.
+ */
+ private Consumer