This is an automated email from the ASF dual-hosted git repository.
ajfabbri pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/trunk by this push:
new 85c23372402 HADOOP-19862: S3A: introduce configurable shared thread
pool for AWS SDK clients (#8426)
85c23372402 is described below
commit 85c233724020e27ca9e5ef69fec2fe6de1d592a1
Author: Konstantin Bereznyakov <[email protected]>
AuthorDate: Tue Jul 7 14:29:21 2026 -0700
HADOOP-19862: S3A: introduce configurable shared thread pool for AWS SDK
clients (#8426)
* S3A: introduce configurable shared thread pool for AWS SDK clients
* missed the need to upgrade new test code to JUnit 5
* addressing addiitional PR feedback (mainly test code)
---
.../src/main/resources/core-default.xml | 119 ++++++
.../org/apache/hadoop/test/GenericTestUtils.java | 22 +
.../java/org/apache/hadoop/fs/s3a/Constants.java | 87 ++++
.../hadoop/fs/s3a/DefaultS3ClientFactory.java | 72 +++-
.../hadoop/fs/s3a/auth/STSClientFactory.java | 30 ++
.../fs/s3a/impl/EncryptionS3ClientFactory.java | 33 ++
.../fs/s3a/impl/LazySharedThreadPoolHolder.java | 142 +++++++
.../fs/s3a/ITestS3ASharedThreadPoolDisabled.java | 96 +++++
.../fs/s3a/ITestS3ASharedThreadPoolEnabled.java | 106 +++++
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 50 +++
.../hadoop/fs/s3a/TestSharedScheduledExecutor.java | 450 +++++++++++++++++++++
11 files changed, 1200 insertions(+), 7 deletions(-)
diff --git
a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index a07460dc308..2609b1d0f4b 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -2258,6 +2258,125 @@ The switch to turn S3A auditing on or off.
</description>
</property>
+<!--
+AWS SDK client shared thread pool settings. When enabled, AWS SDK clients of a
+given type share a single scheduled thread pool instead of each client creating
+its own. This bounds thread growth when many S3A filesystem instances are
+created. Each pool is opt-in and disabled by default.
+
+The default size of 5 matches the per-client default the AWS SDK would
otherwise
+use (Executors.newScheduledThreadPool(5)). Unlike that SDK default, which keeps
+its core threads for the client's lifetime, the shared pool reaps threads that
+are idle beyond the keepalive window.
+-->
+<property>
+ <name>fs.s3a.aws.s3.client.shared.threadpool.enabled</name>
+ <value>false</value>
+ <description>
+ Enable a shared scheduled thread pool for the AWS S3 synchronous client.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.s3.client.shared.threadpool.size</name>
+ <value>5</value>
+ <description>
+ Core pool size of the shared thread pool for the AWS S3 synchronous client.
+ Only used when fs.s3a.aws.s3.client.shared.threadpool.enabled is true.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.s3.client.shared.threadpool.keepalive.seconds</name>
+ <value>60</value>
+ <description>
+ Keepalive time in seconds for idle threads in the shared thread pool for
the
+ AWS S3 synchronous client. Only used when
+ fs.s3a.aws.s3.client.shared.threadpool.enabled is true.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.s3.async.client.shared.threadpool.enabled</name>
+ <value>false</value>
+ <description>
+ Enable a shared scheduled thread pool for the AWS S3 asynchronous client.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.s3.async.client.shared.threadpool.size</name>
+ <value>5</value>
+ <description>
+ Core pool size of the shared thread pool for the AWS S3 asynchronous
client.
+ Only used when fs.s3a.aws.s3.async.client.shared.threadpool.enabled is
true.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.s3.async.client.shared.threadpool.keepalive.seconds</name>
+ <value>60</value>
+ <description>
+ Keepalive time in seconds for idle threads in the shared thread pool for
the
+ AWS S3 asynchronous client. Only used when
+ fs.s3a.aws.s3.async.client.shared.threadpool.enabled is true.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.sts.client.shared.threadpool.enabled</name>
+ <value>false</value>
+ <description>
+ Enable a shared scheduled thread pool for the AWS STS client.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.sts.client.shared.threadpool.size</name>
+ <value>5</value>
+ <description>
+ Core pool size of the shared thread pool for the AWS STS client.
+ Only used when fs.s3a.aws.sts.client.shared.threadpool.enabled is true.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.sts.client.shared.threadpool.keepalive.seconds</name>
+ <value>60</value>
+ <description>
+ Keepalive time in seconds for idle threads in the shared thread pool for
the
+ AWS STS client. Only used when
+ fs.s3a.aws.sts.client.shared.threadpool.enabled is true.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.kms.client.shared.threadpool.enabled</name>
+ <value>false</value>
+ <description>
+ Enable a shared scheduled thread pool for the AWS KMS client.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.kms.client.shared.threadpool.size</name>
+ <value>5</value>
+ <description>
+ Core pool size of the shared thread pool for the AWS KMS client.
+ Only used when fs.s3a.aws.kms.client.shared.threadpool.enabled is true.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.aws.kms.client.shared.threadpool.keepalive.seconds</name>
+ <value>60</value>
+ <description>
+ Keepalive time in seconds for idle threads in the shared thread pool for
the
+ AWS KMS client. Only used when
+ fs.s3a.aws.kms.client.shared.threadpool.enabled is true.
+ </description>
+</property>
+
<!-- Azure file system properties -->
<property>
<name>fs.AbstractFileSystem.wasb.impl</name>
diff --git
a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
index 65c6c8e2cd5..f20a5604978 100644
---
a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
+++
b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
@@ -738,6 +738,28 @@ public static boolean anyThreadMatching(Pattern pattern) {
return false;
}
+ /**
+ * Count the live threads whose name matches the given pattern.
+ * @param pattern a Pattern object used to match thread names
+ * @return the number of live threads whose name matches the pattern
+ */
+ public static int countThreadsMatching(Pattern pattern) {
+ ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
+
+ ThreadInfo[] infos =
+ threadBean.getThreadInfo(threadBean.getAllThreadIds(), 20);
+ int count = 0;
+ for (ThreadInfo info : infos) {
+ if (info == null) {
+ continue;
+ }
+ if (pattern.matcher(info.getThreadName()).matches()) {
+ count++;
+ }
+ }
+ return count;
+ }
+
/**
* Assert that there are no threads running whose name matches the
* given regular expression.
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index 271e9302a3d..a26eefa0179 100644
---
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -1948,4 +1948,91 @@ private Constants() {
* AWS S3 PutObject API Documentation</a>
*/
public static final String IF_NONE_MATCH_STAR = "*";
+
+ // ==================== AWS Client Shared Thread Pool Configuration
===========
+ // These settings control whether AWS SDK clients share a thread pool
+ // instead of each client creating its own. This prevents thread leaks
+ // when many S3A filesystem instances are created.
+
+ /**
+ * Default thread pool size for AWS client shared thread pools: {@value}.
+ */
+ public static final int AWS_CLIENT_SHARED_THREADPOOL_SIZE_DEFAULT = 5;
+
+ /**
+ * Default keepalive in seconds for AWS client shared thread pool: {@value}.
+ */
+ public static final int AWS_CLIENT_SHARED_THREADPOOL_KEEPALIVE_DEFAULT = 60;
+
+ /**
+ * Enable shared thread pool for AWS S3 sync client: {@value}.
+ */
+ public static final String AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED =
+ "fs.s3a.aws.s3.client.shared.threadpool.enabled";
+
+ /**
+ * Thread pool size for AWS S3 sync client shared thread pool: {@value}.
+ */
+ public static final String AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE =
+ "fs.s3a.aws.s3.client.shared.threadpool.size";
+
+ /**
+ * Keepalive in seconds for AWS S3 sync client shared thread pool: {@value}.
+ */
+ public static final String AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE =
+ "fs.s3a.aws.s3.client.shared.threadpool.keepalive.seconds";
+
+ /**
+ * Enable shared thread pool for AWS S3 async client: {@value}.
+ */
+ public static final String AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED =
+ "fs.s3a.aws.s3.async.client.shared.threadpool.enabled";
+
+ /**
+ * Thread pool size for AWS S3 async client shared thread pool: {@value}.
+ */
+ public static final String AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE =
+ "fs.s3a.aws.s3.async.client.shared.threadpool.size";
+
+ /**
+ * Keepalive in seconds for AWS S3 async client shared thread pool: {@value}.
+ */
+ public static final String AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_KEEPALIVE =
+ "fs.s3a.aws.s3.async.client.shared.threadpool.keepalive.seconds";
+
+ /**
+ * Enable shared thread pool for AWS STS client: {@value}.
+ */
+ public static final String AWS_STS_CLIENT_SHARED_THREADPOOL_ENABLED =
+ "fs.s3a.aws.sts.client.shared.threadpool.enabled";
+
+ /**
+ * Thread pool size for AWS STS client shared thread pool: {@value}.
+ */
+ public static final String AWS_STS_CLIENT_SHARED_THREADPOOL_SIZE =
+ "fs.s3a.aws.sts.client.shared.threadpool.size";
+
+ /**
+ * Keepalive in seconds for AWS STS client shared thread pool: {@value}.
+ */
+ public static final String AWS_STS_CLIENT_SHARED_THREADPOOL_KEEPALIVE =
+ "fs.s3a.aws.sts.client.shared.threadpool.keepalive.seconds";
+
+ /**
+ * Enable shared thread pool for AWS KMS client: {@value}.
+ */
+ public static final String AWS_KMS_CLIENT_SHARED_THREADPOOL_ENABLED =
+ "fs.s3a.aws.kms.client.shared.threadpool.enabled";
+
+ /**
+ * Thread pool size for AWS KMS client shared thread pool: {@value}.
+ */
+ public static final String AWS_KMS_CLIENT_SHARED_THREADPOOL_SIZE =
+ "fs.s3a.aws.kms.client.shared.threadpool.size";
+
+ /**
+ * Keepalive in seconds for AWS KMS client shared thread pool: {@value}.
+ */
+ public static final String AWS_KMS_CLIENT_SHARED_THREADPOOL_KEEPALIVE =
+ "fs.s3a.aws.kms.client.shared.threadpool.keepalive.seconds";
}
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
index 41e904ec9de..c8446c38c24 100644
---
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
@@ -21,11 +21,13 @@
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.fs.s3a.impl.AWSClientConfig;
+import org.apache.hadoop.fs.s3a.impl.LazySharedThreadPoolHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,17 +65,23 @@
import static org.apache.hadoop.fs.s3a.Constants.AWS_REGION;
import static org.apache.hadoop.fs.s3a.Constants.AWS_S3_ACCESS_GRANTS_ENABLED;
import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE;
import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CROSS_REGION_ACCESS_ENABLED;
import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CROSS_REGION_ACCESS_ENABLED_DEFAULT;
import static org.apache.hadoop.fs.s3a.Constants.AWS_S3_DEFAULT_REGION;
+import static org.apache.hadoop.fs.s3a.Constants.AWS_SERVICE_IDENTIFIER_S3;
import static org.apache.hadoop.fs.s3a.Constants.CENTRAL_ENDPOINT;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_SECURE_CONNECTIONS;
import static org.apache.hadoop.fs.s3a.Constants.FIPS_ENDPOINT;
import static org.apache.hadoop.fs.s3a.Constants.HTTP_SIGNER_CLASS_NAME;
import static org.apache.hadoop.fs.s3a.Constants.HTTP_SIGNER_ENABLED;
import static org.apache.hadoop.fs.s3a.Constants.HTTP_SIGNER_ENABLED_DEFAULT;
-import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_SECURE_CONNECTIONS;
import static org.apache.hadoop.fs.s3a.Constants.SECURE_CONNECTIONS;
-import static org.apache.hadoop.fs.s3a.Constants.AWS_SERVICE_IDENTIFIER_S3;
import static org.apache.hadoop.fs.s3a.auth.SignerFactory.createHttpSigner;
import static org.apache.hadoop.fs.s3a.impl.AWSHeaders.REQUESTER_PAYS_HEADER;
import static
org.apache.hadoop.fs.s3a.impl.InternalConstants.AUTH_SCHEME_AWS_SIGV_4;
@@ -97,6 +105,46 @@ public class DefaultS3ClientFactory extends Configured
private static final Pattern VPC_ENDPOINT_PATTERN =
Pattern.compile("^(?:.+\\.)?([a-z0-9-]+)\\.vpce\\.amazonaws\\.(?:com|com\\.cn)$");
+ /**
+ * Shared executor for S3 sync clients.
+ */
+ private static final LazySharedThreadPoolHolder S3_SYNC_EXECUTOR =
+ new LazySharedThreadPoolHolder(
+ AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ "s3a-s3-sync-scheduler");
+
+ /**
+ * Shared executor for S3 async clients.
+ */
+ private static final LazySharedThreadPoolHolder S3_ASYNC_EXECUTOR =
+ new LazySharedThreadPoolHolder(
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ "s3a-s3-async-scheduler");
+
+ /**
+ * Get the shared executor holder for S3 sync clients.
+ * This is for testing only.
+ * @return the holder
+ */
+ @VisibleForTesting
+ static LazySharedThreadPoolHolder s3SyncExecutorHolder() {
+ return S3_SYNC_EXECUTOR;
+ }
+
+ /**
+ * Get the shared executor holder for S3 async clients.
+ * This is for testing only.
+ * @return the holder
+ */
+ @VisibleForTesting
+ static LazySharedThreadPoolHolder s3AsyncExecutorHolder() {
+ return S3_ASYNC_EXECUTOR;
+ }
+
/**
* Subclasses refer to this.
*/
@@ -235,8 +283,10 @@ private <BuilderT extends S3BaseClientBuilder<BuilderT,
ClientT>, ClientT> Build
.pathStyleAccessEnabled(parameters.isPathStyleAccess())
.build();
- final ClientOverrideConfiguration.Builder override =
- createClientOverrideConfiguration(parameters, conf);
+ final ClientOverrideConfiguration.Builder override =
createClientOverrideConfiguration(
+ parameters,
+ conf,
+ builder instanceof S3AsyncClientBuilder);
S3BaseClientBuilder<BuilderT, ClientT> s3BaseClientBuilder = builder
.overrideConfiguration(override.build())
@@ -265,13 +315,14 @@ private <BuilderT extends S3BaseClientBuilder<BuilderT,
ClientT>, ClientT> Build
* Create an override configuration for an S3 client.
* @param parameters parameter object
* @param conf configuration object
- * @throws IOException any IOE raised, or translated exception
- * @throws RuntimeException some failures creating an http signer
+ * @param isAsync true for async client, false for sync client
* @return the override configuration
* @throws IOException any IOE raised, or translated exception
+ * @throws RuntimeException some failures creating an http signer
*/
protected ClientOverrideConfiguration.Builder
createClientOverrideConfiguration(
- S3ClientCreationParameters parameters, Configuration conf) throws
IOException {
+ S3ClientCreationParameters parameters, Configuration conf, boolean
isAsync)
+ throws IOException {
final ClientOverrideConfiguration.Builder clientOverrideConfigBuilder =
AWSClientConfig.createClientConfigBuilder(conf,
AWS_SERVICE_IDENTIFIER_S3);
@@ -302,6 +353,13 @@ protected ClientOverrideConfiguration.Builder
createClientOverrideConfiguration(
final RetryPolicy.Builder retryPolicyBuilder =
AWSClientConfig.createRetryPolicyBuilder(conf);
clientOverrideConfigBuilder.retryPolicy(retryPolicyBuilder.build());
+ ScheduledExecutorService executor = isAsync
+ ? S3_ASYNC_EXECUTOR.get(conf)
+ : S3_SYNC_EXECUTOR.get(conf);
+ if (executor != null) {
+ clientOverrideConfigBuilder.scheduledExecutorService(executor);
+ }
+
return clientOverrideConfigBuilder;
}
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/STSClientFactory.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/STSClientFactory.java
index ab151744969..56d9056f920 100644
---
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/STSClientFactory.java
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/STSClientFactory.java
@@ -22,6 +22,7 @@
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
@@ -35,6 +36,7 @@
import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
import software.amazon.awssdk.services.sts.model.Credentials;
import software.amazon.awssdk.services.sts.model.GetSessionTokenRequest;
+import org.apache.hadoop.fs.s3a.impl.LazySharedThreadPoolHolder;
import org.apache.hadoop.fs.s3a.impl.AWSClientConfig;
import org.apache.hadoop.util.Preconditions;
@@ -44,6 +46,7 @@
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.s3a.Invoker;
import org.apache.hadoop.fs.s3a.Retries;
@@ -51,6 +54,9 @@
import static org.apache.commons.lang3.StringUtils.isEmpty;
import static org.apache.commons.lang3.StringUtils.isNotEmpty;
import static org.apache.hadoop.fs.s3a.Constants.AWS_SERVICE_IDENTIFIER_STS;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_STS_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_STS_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_STS_CLIENT_SHARED_THREADPOOL_SIZE;
import static org.apache.hadoop.fs.s3a.auth.delegation.DelegationConstants.*;
/**
@@ -63,6 +69,26 @@ public class STSClientFactory {
private static final Logger LOG =
LoggerFactory.getLogger(STSClientFactory.class);
+ /**
+ * Shared executor for STS clients.
+ */
+ private static final LazySharedThreadPoolHolder STS_EXECUTOR =
+ new LazySharedThreadPoolHolder(
+ AWS_STS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ "s3a-sts-scheduler");
+
+ /**
+ * Get the shared executor holder for STS clients.
+ * This is for testing only.
+ * @return the holder
+ */
+ @VisibleForTesting
+ public static LazySharedThreadPoolHolder stsExecutorHolder() {
+ return STS_EXECUTOR;
+ }
+
/**
* Create the builder ready for any final configuration options.
* Picks up connection settings from the Hadoop configuration, including
@@ -139,6 +165,10 @@ public static StsClientBuilder builder(final
AwsCredentialsProvider credentials,
final ProxyConfiguration proxyConfig =
AWSClientConfig.createProxyConfiguration(conf, bucket);
clientOverrideConfigBuilder.retryPolicy(retryPolicyBuilder.build());
+ ScheduledExecutorService executor = STS_EXECUTOR.get(conf);
+ if (executor != null) {
+ clientOverrideConfigBuilder.scheduledExecutorService(executor);
+ }
httpClientBuilder.proxyConfiguration(proxyConfig);
stsClientBuilder.httpClientBuilder(httpClientBuilder)
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/EncryptionS3ClientFactory.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/EncryptionS3ClientFactory.java
index 3c2756dfb05..381d560d367 100644
---
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/EncryptionS3ClientFactory.java
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/EncryptionS3ClientFactory.java
@@ -20,7 +20,9 @@
import java.io.IOException;
import java.net.URI;
+import java.util.concurrent.ScheduledExecutorService;
+import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kms.KmsClient;
import software.amazon.awssdk.services.kms.KmsClientBuilder;
@@ -33,12 +35,16 @@
import software.amazon.encryption.s3.materials.Keyring;
import software.amazon.encryption.s3.materials.KmsKeyring;
+import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.s3a.DefaultS3ClientFactory;
import org.apache.hadoop.util.Preconditions;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.functional.LazyAtomicReference;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_KMS_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_KMS_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_KMS_CLIENT_SHARED_THREADPOOL_SIZE;
import static
org.apache.hadoop.fs.s3a.impl.InstantiationIOException.unavailable;
/**
@@ -46,6 +52,26 @@
*/
public class EncryptionS3ClientFactory extends DefaultS3ClientFactory {
+ /**
+ * Shared executor for KMS clients.
+ */
+ private static final LazySharedThreadPoolHolder KMS_EXECUTOR =
+ new LazySharedThreadPoolHolder(
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ "s3a-kms-scheduler");
+
+ /**
+ * Get the shared executor holder for KMS clients.
+ * This is for testing only.
+ * @return the holder
+ */
+ @VisibleForTesting
+ public static LazySharedThreadPoolHolder kmsExecutorHolder() {
+ return KMS_EXECUTOR;
+ }
+
/**
* Encryption client class name.
* value: {@value}
@@ -201,6 +227,13 @@ private S3Client
createS3EncryptionClient(S3ClientCreationParameters parameters)
private Keyring createKmsKeyring(S3ClientCreationParameters parameters,
CSEMaterials cseMaterials) {
KmsClientBuilder kmsClientBuilder = KmsClient.builder();
+ ScheduledExecutorService executor =
KMS_EXECUTOR.get(cseMaterials.getConf());
+ if (executor != null) {
+ kmsClientBuilder.overrideConfiguration(
+ ClientOverrideConfiguration.builder()
+ .scheduledExecutorService(executor)
+ .build());
+ }
if (parameters.getCredentialSet() != null) {
kmsClientBuilder.credentialsProvider(parameters.getCredentialSet());
}
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/LazySharedThreadPoolHolder.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/LazySharedThreadPoolHolder.java
new file mode 100644
index 00000000000..acae2fecfc8
--- /dev/null
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/LazySharedThreadPoolHolder.java
@@ -0,0 +1,142 @@
+/*
+ * 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.hadoop.fs.s3a.impl;
+
+import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.conf.Configuration;
+
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_CLIENT_SHARED_THREADPOOL_KEEPALIVE_DEFAULT;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_CLIENT_SHARED_THREADPOOL_SIZE_DEFAULT;
+import static org.apache.hadoop.fs.s3a.S3AUtils.intOption;
+
+/**
+ * Holder for a lazily initialized shared ScheduledExecutorService.
+ */
[email protected]
[email protected]
+public class LazySharedThreadPoolHolder {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(LazySharedThreadPoolHolder.class);
+
+ private final String enabledKey;
+ private final String sizeKey;
+ private final String keepAliveKey;
+ private final String namePrefix;
+
+ private Optional<ScheduledExecutorService> executor;
+
+ /**
+ * Create a holder for a lazy shared thread pool.
+ * @param enabledKey config key to enable the shared pool
+ * @param sizeKey config key for pool size
+ * @param keepAliveKey config key for thread keep-alive in seconds
+ * @param namePrefix thread name prefix for debugging
+ */
+ public LazySharedThreadPoolHolder(String enabledKey, String sizeKey,
+ String keepAliveKey, String namePrefix) {
+ this.enabledKey = enabledKey;
+ this.sizeKey = sizeKey;
+ this.keepAliveKey = keepAliveKey;
+ this.namePrefix = namePrefix;
+ }
+
+ /**
+ * Get the config key controlling whether the shared pool is enabled.
+ * This is for testing only.
+ * @return the enabled config key
+ */
+ @VisibleForTesting
+ public String getEnabledKey() {
+ return enabledKey;
+ }
+
+ /**
+ * Get the config key controlling the pool size.
+ * This is for testing only.
+ * @return the size config key
+ */
+ @VisibleForTesting
+ public String getSizeKey() {
+ return sizeKey;
+ }
+
+ /**
+ * Get the config key controlling the thread keep-alive.
+ * This is for testing only.
+ * @return the keep-alive config key
+ */
+ @VisibleForTesting
+ public String getKeepAliveKey() {
+ return keepAliveKey;
+ }
+
+ /**
+ * Get the shared executor, creating it on first call if enabled.
+ * @param conf configuration
+ * @return the executor, or null if not enabled
+ */
+ public synchronized ScheduledExecutorService get(Configuration conf) {
+ if (executor == null) {
+ if (conf.getBoolean(enabledKey, false)) {
+ int poolSize = intOption(conf, sizeKey,
+ AWS_CLIENT_SHARED_THREADPOOL_SIZE_DEFAULT, 1);
+ int keepAlive = intOption(conf, keepAliveKey,
+ AWS_CLIENT_SHARED_THREADPOOL_KEEPALIVE_DEFAULT, 1);
+ executor = Optional.of(createScheduledExecutor(namePrefix, poolSize,
keepAlive));
+ } else {
+ executor = Optional.empty();
+ }
+ }
+ return executor.orElse(null);
+ }
+
+ /**
+ * Create a scheduled executor with idle thread timeout.
+ * @param namePrefix thread name prefix for debugging
+ * @param poolSize core pool size
+ * @param keepAliveSeconds keepalive time in seconds
+ * @return the executor
+ */
+ @VisibleForTesting
+ public static ScheduledExecutorService createScheduledExecutor(
+ String namePrefix, int poolSize, int keepAliveSeconds) {
+ ScheduledThreadPoolExecutor pool = new
ScheduledThreadPoolExecutor(poolSize,
+ runnable -> {
+ Thread t = new Thread(runnable, namePrefix);
+ t.setDaemon(true);
+ return t;
+ });
+ pool.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
+ pool.allowCoreThreadTimeOut(true);
+ LOG.debug("Created shared executor '{}' with pool size {} and keepalive
{}s",
+ namePrefix, poolSize, keepAliveSeconds);
+ return pool;
+ }
+}
diff --git
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolDisabled.java
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolDisabled.java
new file mode 100644
index 00000000000..d78c7135b59
--- /dev/null
+++
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolDisabled.java
@@ -0,0 +1,96 @@
+/*
+ * 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.hadoop.fs.s3a;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED;
+
+/**
+ * Control case for {@link ITestS3ASharedThreadPoolEnabled}: with the shared
+ * pool disabled (the default), each S3AFileSystem's AWS SDK clients create
their
+ * own sdk-ScheduledExecutor pool, so thread count grows with the number of
+ * instances. Must run in its own JVM: the static pool holder is single-state
+ * per JVM, so this control (pools off) and {@link
ITestS3ASharedThreadPoolEnabled}
+ * (pools on) cannot share a fork. hadoop-aws sets reuseForks=false, which
these
+ * tests require; do not enable fork reuse for them.
+ * <p>
+ * The sdk-ScheduledExecutor pools are created lazily as the clients schedule
+ * work, so the exact count varies; the assertion is only that it clearly
+ * exceeds a single shared pool.
+ */
+public class ITestS3ASharedThreadPoolDisabled extends AbstractS3ATestBase {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ITestS3ASharedThreadPoolDisabled.class);
+
+ private static final int INSTANCES = 20;
+
+ /** Pool size the shared pools would otherwise cap at, per type. */
+ private static final int SHARED_POOL_SIZE = 5;
+
+ /** Thread name prefix the AWS SDK uses for its per-client default pool. */
+ private static final String SDK_POOL_PREFIX = "sdk-ScheduledExecutor";
+
+ @Override
+ protected Configuration createConfiguration() {
+ Configuration conf = super.createConfiguration();
+ // Strip any base/per-bucket override, then pin the pools explicitly off,
so
+ // this control is deterministic regardless of the default: it must run
with
+ // the shared pools disabled to exercise the per-client leak.
+ S3ATestUtils.removeBaseAndBucketOverrides(
+ S3ATestUtils.getTestBucketName(conf), conf,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED);
+ conf.setBoolean(AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED, false);
+ conf.setBoolean(AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED, false);
+ return conf;
+ }
+
+ @Test
+ public void testWithoutSharedPoolThreadsLeakPastGc() throws Exception {
+ // Abandon many uncached instances and force GC: without sharing, each
+ // client's own scheduled pool is leaked (not shut down on eviction, and
+ // rooted by its own running threads), so the threads survive collection.
+ int sdkThreads =
S3ATestUtils.countSchedulerThreadsAfterAbandoningFilesystems(
+ getFileSystem().getUri(), getConfiguration(), INSTANCES,
+ SDK_POOL_PREFIX);
+ LOG.info("SDK scheduler threads surviving GC after abandoning {} "
+ + "S3AFileSystem instances without the shared pool: {}", INSTANCES,
+ sdkThreads);
+
+ // The leaked per-client pools persist past GC, far exceeding a single
+ // shared pool.
+ Assertions.assertThat(sdkThreads)
+ .as("Without the shared pool, abandoned sdk-ScheduledExecutor threads "
+ + "should leak past GC, exceeding a single shared pool after %s "
+ + "instances", INSTANCES)
+ .isGreaterThan(2 * SHARED_POOL_SIZE);
+
+ // Intentionally not cleaned up: the abandoned uncached instances leave no
+ // handle to close, and that unreclaimability is the point. The daemon
+ // threads are reaped when this class's own fork exits (reuseForks=false).
+ }
+}
diff --git
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolEnabled.java
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolEnabled.java
new file mode 100644
index 00000000000..e7406f7d910
--- /dev/null
+++
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolEnabled.java
@@ -0,0 +1,106 @@
+/*
+ * 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.hadoop.fs.s3a;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE;
+
+/**
+ * Integration test demonstrating that the shared scheduled thread pool bounds
+ * thread growth across many S3AFileSystem instances.
+ * <p>
+ * Background (HADOOP-19862): each S3AFileSystem builds its own AWS SDK
clients,
+ * and each client otherwise creates a 5-thread sdk-ScheduledExecutor pool, so
+ * creating many instances grows threads without bound. With the shared pool
+ * enabled, all clients of a type reuse a single pool, so the thread count
stays
+ * bounded regardless of how many instances are created.
+ * <p>
+ * This test proves the fix direction (pool enabled, threads bounded);
+ * {@link ITestS3ASharedThreadPoolDisabled} is the control proving the opposite
+ * (pool disabled, threads grow). The two must be separate classes: the holders
+ * are private static final and memoize on first use, so the first
configuration
+ * to reach them wins for the JVM. They therefore cannot share a fork; each
must
+ * run in its own JVM. hadoop-aws sets reuseForks=false, which these tests
+ * require (do not enable fork reuse for them); do not merge them into one
class.
+ * <p>
+ * Exact thread counts depend on AWS SDK internals, so the assertion is on the
+ * bound (not exceeding the pool sizes), not an exact number.
+ */
+public class ITestS3ASharedThreadPoolEnabled extends AbstractS3ATestBase {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ITestS3ASharedThreadPoolEnabled.class);
+
+ private static final int POOL_SIZE = 5;
+
+ /** Number of distinct filesystem instances to create. */
+ private static final int INSTANCES = 20;
+
+ /** Thread name prefixes used by the shared pools (see the factories). */
+ private static final String[] SHARED_POOL_PREFIXES = {
+ "s3a-s3-sync-scheduler",
+ "s3a-s3-async-scheduler",
+ };
+
+ @Override
+ protected Configuration createConfiguration() {
+ Configuration conf = super.createConfiguration();
+ // Strip any base or per-bucket overrides for these keys so the values set
+ // below win: a bucket override would otherwise take precedence and
+ // silently invalidate the test.
+ S3ATestUtils.removeBaseAndBucketOverrides(
+ S3ATestUtils.getTestBucketName(conf), conf,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE);
+ // Enable the shared pools before any client is built, so the static
+ // holders memoize in the enabled state for this JVM.
+ conf.setBoolean(AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED, true);
+ conf.setBoolean(AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED, true);
+ conf.setInt(AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE, POOL_SIZE);
+ conf.setInt(AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE, POOL_SIZE);
+ return conf;
+ }
+
+ @Test
+ public void testSharedPoolBoundsThreadsAcrossManyInstances() throws
Exception {
+ int shared = S3ATestUtils.countSchedulerThreadsAfterAbandoningFilesystems(
+ getFileSystem().getUri(), getConfiguration(), INSTANCES,
+ SHARED_POOL_PREFIXES);
+ LOG.info("Shared scheduler threads after abandoning {} S3AFileSystem "
+ + "instances and forcing GC: {}", INSTANCES, shared);
+
+ // With sharing, the single static pool survives instance abandonment + GC
+ // and stays bounded by the two pools' core sizes, regardless of count.
+ Assertions.assertThat(shared)
+ .as("Total shared scheduler threads after abandoning %s S3AFileSystem "
+ + "instances should be bounded by the pool sizes", INSTANCES)
+ .isLessThanOrEqualTo(2 * POOL_SIZE);
+ }
+}
diff --git
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
index e5205d64854..cf5c6b45b4b 100644
---
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
+++
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java
@@ -87,6 +87,7 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
+import java.lang.ref.WeakReference;
import java.lang.reflect.Field;
import java.net.URI;
import java.net.URISyntaxException;
@@ -101,6 +102,8 @@
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.regex.Pattern;
import java.util.stream.Collectors;
import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
@@ -115,6 +118,8 @@
import static
org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils.getExternalData;
import static
org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils.requireDefaultExternalDataFile;
import static org.apache.hadoop.test.GenericTestUtils.buildPaths;
+import static org.apache.hadoop.test.GenericTestUtils.countThreadsMatching;
+import static org.apache.hadoop.test.GenericTestUtils.waitFor;
import static org.apache.hadoop.util.Preconditions.checkNotNull;
import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_CREDENTIAL_PROVIDER_PATH;
import static org.apache.commons.lang3.StringUtils.isNotEmpty;
@@ -2052,4 +2057,49 @@ public static Pair<Long, Long> requestRange(String
rangeHeader) {
// error case
return Pair.of(-1L, -1L);
}
+
+ /**
+ * Count live threads (by name prefix) that survive collection of the
+ * {@code S3AFileSystem} instances owning them. Creates {@code instances}
+ * uncached filesystems, lists on each to spawn its SDK scheduler threads,
+ * then abandons them and requests GC until a {@link WeakReference} to one
+ * clears — proving an owner was really collected, as {@link System#gc()} is
+ * only a hint. Per aws-sdk-java-v2 #1690 a leaked per-client scheduler's
+ * running threads root its executor, so it outlives its owner.
+ * @param uri filesystem URI
+ * @param conf base configuration (filesystem caching is forced off here)
+ * @param instances number of filesystem instances to create and abandon
+ * @param prefixes scheduler thread name prefixes to count
+ * @return number of matching live threads remaining after the owners are
collected
+ * @throws IOException creating a filesystem or listing failed
+ * @throws InterruptedException interrupted while waiting for collection
+ * @throws TimeoutException an abandoned instance was not collected in time
+ */
+ public static int countSchedulerThreadsAfterAbandoningFilesystems(URI uri,
+ Configuration conf, int instances, String... prefixes)
+ throws IOException, InterruptedException, TimeoutException {
+ Configuration uncached = new Configuration(conf);
+ disableFilesystemCaching(uncached);
+ WeakReference<FileSystem> sentinel = null;
+ for (int i = 0; i < instances; i++) {
+ // Uncached instance; deliberately neither retained nor closed, so it
+ // becomes eligible for collection once this iteration ends.
+ FileSystem fs = FileSystem.get(uri, uncached);
+ fs.listStatus(new Path("/"));
+ if (sentinel == null) {
+ sentinel = new WeakReference<>(fs);
+ }
+ }
+ // System.gc() is only a hint, so prove a real collection happened: request
+ // GC until the weak reference to an abandoned instance clears.
+ final WeakReference<FileSystem> ref = sentinel;
+ waitFor(() -> {
+ System.gc();
+ return ref.get() == null;
+ }, 100, 30_000);
+ Pattern pattern = Pattern.compile(Arrays.stream(prefixes)
+ .map(p -> Pattern.quote(p) + ".*")
+ .collect(Collectors.joining("|")));
+ return countThreadsMatching(pattern);
+ }
}
diff --git
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestSharedScheduledExecutor.java
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestSharedScheduledExecutor.java
new file mode 100644
index 00000000000..f302fc177b7
--- /dev/null
+++
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestSharedScheduledExecutor.java
@@ -0,0 +1,450 @@
+/*
+ * 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.hadoop.fs.s3a;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.auth.STSClientFactory;
+import org.apache.hadoop.fs.s3a.impl.EncryptionS3ClientFactory;
+import org.apache.hadoop.fs.s3a.impl.LazySharedThreadPoolHolder;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_CLIENT_SHARED_THREADPOOL_KEEPALIVE_DEFAULT;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_CLIENT_SHARED_THREADPOOL_SIZE_DEFAULT;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_KMS_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_KMS_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_KMS_CLIENT_SHARED_THREADPOOL_SIZE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_STS_CLIENT_SHARED_THREADPOOL_ENABLED;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_STS_CLIENT_SHARED_THREADPOOL_KEEPALIVE;
+import static
org.apache.hadoop.fs.s3a.Constants.AWS_STS_CLIENT_SHARED_THREADPOOL_SIZE;
+
+/**
+ * Tests for the shared scheduled executors in DefaultS3ClientFactory.
+ */
+public class TestSharedScheduledExecutor {
+
+ @Test
+ public void testLazyHolderDisabledByDefault() {
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ "test.enabled", "test.size", "test.keepalive", "test-disabled");
+ Configuration conf = new Configuration();
+ ScheduledExecutorService executor = holder.get(conf);
+ Assertions.assertThat(executor)
+ .as("Executor should be null when disabled")
+ .isNull();
+ }
+
+ @Test
+ public void testCreateScheduledExecutorConfiguration() {
+ ScheduledExecutorService executor =
+ LazySharedThreadPoolHolder.createScheduledExecutor("test-scheduler",
10, 30);
+ Assertions.assertThat(executor)
+ .as("Executor should be a ScheduledThreadPoolExecutor")
+ .isInstanceOf(ScheduledThreadPoolExecutor.class);
+
+ ScheduledThreadPoolExecutor poolExecutor = (ScheduledThreadPoolExecutor)
executor;
+ Assertions.assertThat(poolExecutor.getCorePoolSize())
+ .as("Core pool size should be 10")
+ .isEqualTo(10);
+ Assertions.assertThat(poolExecutor.getKeepAliveTime(TimeUnit.SECONDS))
+ .as("Keepalive should be 30 seconds")
+ .isEqualTo(30);
+ Assertions.assertThat(poolExecutor.allowsCoreThreadTimeOut())
+ .as("Core threads should be allowed to time out")
+ .isTrue();
+
+ executor.shutdown();
+ }
+
+ @Test
+ public void testCreateScheduledExecutorThreadsAreDaemon() throws Exception {
+ ScheduledExecutorService executor =
+ LazySharedThreadPoolHolder.createScheduledExecutor("test-daemon", 5,
60);
+ final boolean[] isDaemon = new boolean[1];
+ executor.submit(() -> {
+ isDaemon[0] = Thread.currentThread().isDaemon();
+ }).get();
+ Assertions.assertThat(isDaemon[0])
+ .as("Executor threads should be daemon threads")
+ .isTrue();
+ executor.shutdown();
+ }
+
+ @Test
+ public void testCreateScheduledExecutorThreadName() throws Exception {
+ ScheduledExecutorService executor =
+ LazySharedThreadPoolHolder.createScheduledExecutor("custom-prefix", 5,
60);
+ final String[] threadName = new String[1];
+ executor.submit(() -> {
+ threadName[0] = Thread.currentThread().getName();
+ }).get();
+ Assertions.assertThat(threadName[0])
+ .as("Thread name should match custom prefix")
+ .startsWith("custom-prefix");
+ executor.shutdown();
+ }
+
+ @Test
+ public void testLazyHolderRejectsNegativePoolSize() {
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ "test.enabled", "test.size", "test.keepalive", "test-pool");
+ Configuration conf = new Configuration();
+ conf.setBoolean("test.enabled", true);
+ conf.setInt("test.size", -1);
+ conf.setInt("test.keepalive", 60);
+ Assertions.assertThatThrownBy(() -> holder.get(conf))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("test.size")
+ .hasMessageContaining("below the minimum value");
+ }
+
+ @Test
+ public void testLazyHolderRejectsZeroPoolSize() {
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ "test.enabled", "test.size", "test.keepalive", "test-pool");
+ Configuration conf = new Configuration();
+ conf.setBoolean("test.enabled", true);
+ conf.setInt("test.size", 0);
+ conf.setInt("test.keepalive", 60);
+ Assertions.assertThatThrownBy(() -> holder.get(conf))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("test.size")
+ .hasMessageContaining("below the minimum value");
+ }
+
+ @Test
+ public void testLazyHolderRejectsNegativeKeepAlive() {
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ "test.enabled", "test.size", "test.keepalive", "test-pool");
+ Configuration conf = new Configuration();
+ conf.setBoolean("test.enabled", true);
+ conf.setInt("test.size", 5);
+ conf.setInt("test.keepalive", -1);
+ Assertions.assertThatThrownBy(() -> holder.get(conf))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("test.keepalive")
+ .hasMessageContaining("below the minimum value");
+ }
+
+ @Test
+ public void testLazyHolderRejectsZeroKeepAlive() {
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ "test.enabled", "test.size", "test.keepalive", "test-pool");
+ Configuration conf = new Configuration();
+ conf.setBoolean("test.enabled", true);
+ conf.setInt("test.size", 5);
+ conf.setInt("test.keepalive", 0);
+ Assertions.assertThatThrownBy(() -> holder.get(conf))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("test.keepalive")
+ .hasMessageContaining("below the minimum value");
+ }
+
+ @Test
+ public void testLazyHolderAcceptsValidConfig() {
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ "test.enabled", "test.size", "test.keepalive", "test-valid");
+ Configuration conf = new Configuration();
+ conf.setBoolean("test.enabled", true);
+ conf.setInt("test.size", 5);
+ conf.setInt("test.keepalive", 30);
+ ScheduledExecutorService executor = holder.get(conf);
+ Assertions.assertThat(executor)
+ .as("Executor should be created with valid config")
+ .isNotNull()
+ .isInstanceOf(ScheduledThreadPoolExecutor.class);
+
+ ScheduledThreadPoolExecutor poolExecutor =
+ (ScheduledThreadPoolExecutor) executor;
+ Assertions.assertThat(poolExecutor.getCorePoolSize())
+ .as("Core pool size should match configured value")
+ .isEqualTo(5);
+ Assertions.assertThat(poolExecutor.getKeepAliveTime(TimeUnit.SECONDS))
+ .as("Keepalive should match configured value")
+ .isEqualTo(30);
+ executor.shutdown();
+ }
+
+ /**
+ * Count threads matching the given prefix.
+ * @param prefix thread name prefix to match
+ * @return count of matching threads
+ */
+ private int countThreadsWithPrefix(String prefix) {
+ return GenericTestUtils.countThreadsMatching(
+ Pattern.compile(Pattern.quote(prefix) + ".*"));
+ }
+
+ /**
+ * Test that without shared pool, each holder creates its own threads.
+ * This demonstrates the thread growth problem.
+ */
+ @Test
+ public void testWithoutSharedPoolThreadsGrow() throws Exception {
+ final String prefix = "test-growth-";
+ final int poolSize = 3;
+ final int numHolders = 5;
+ List<ScheduledExecutorService> executors = new ArrayList<>();
+
+ int initialCount = countThreadsWithPrefix(prefix);
+
+ for (int i = 0; i < numHolders; i++) {
+ ScheduledExecutorService executor =
+ LazySharedThreadPoolHolder.createScheduledExecutor(prefix + i,
poolSize, 60);
+ executors.add(executor);
+ executor.submit(() -> {}).get();
+ }
+
+ int afterCount = countThreadsWithPrefix(prefix);
+ int newThreads = afterCount - initialCount;
+
+ Assertions.assertThat(newThreads)
+ .as("Without shared pool, thread count should grow with each executor")
+ .isGreaterThanOrEqualTo(numHolders);
+
+ for (ScheduledExecutorService executor : executors) {
+ executor.shutdown();
+ }
+ }
+
+ /**
+ * Test that with shared pool enabled, thread count is bounded.
+ * This demonstrates the fix for thread leak.
+ */
+ @Test
+ public void testWithSharedPoolThreadCountBounded() throws Exception {
+ final String prefix = "test-shared-";
+ final int poolSize = 5;
+ final int numCalls = 10;
+
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ prefix);
+
+ Configuration conf = new Configuration();
+ conf.setBoolean(AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED, true);
+ conf.setInt(AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE, poolSize);
+ conf.setInt(AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE, 60);
+
+ int initialCount = countThreadsWithPrefix(prefix);
+
+ for (int i = 0; i < numCalls; i++) {
+ ScheduledExecutorService executor = holder.get(conf);
+ Assertions.assertThat(executor)
+ .as("Should return same executor instance")
+ .isNotNull();
+ executor.submit(() -> {}).get();
+ }
+
+ int afterCount = countThreadsWithPrefix(prefix);
+ int newThreads = afterCount - initialCount;
+
+ Assertions.assertThat(newThreads)
+ .as("With shared pool, thread count should be bounded by pool size")
+ .isLessThanOrEqualTo(poolSize);
+
+ holder.get(conf).shutdown();
+ }
+
+ /**
+ * Test that holder returns the same executor instance on repeated calls.
+ */
+ @Test
+ public void testHolderReturnsSameInstance() {
+ LazySharedThreadPoolHolder holder = new LazySharedThreadPoolHolder(
+ "test.enabled", "test.size", "test.keepalive", "test-same");
+ Configuration conf = new Configuration();
+ conf.setBoolean("test.enabled", true);
+ conf.setInt("test.size", 5);
+ conf.setInt("test.keepalive", 60);
+
+ ScheduledExecutorService first = holder.get(conf);
+ ScheduledExecutorService second = holder.get(conf);
+ ScheduledExecutorService third = holder.get(conf);
+
+ Assertions.assertThat(first)
+ .as("Holder should return same instance on repeated calls")
+ .isSameAs(second)
+ .isSameAs(third);
+
+ first.shutdown();
+ }
+
+ /**
+ * Verify the production shared-thread-pool config keys for every AWS client
+ * type are well-formed: within each client the enabled/size/keepalive keys
+ * share a single base and differ only by the expected suffix, and all keys
+ * are globally unique. This guards against the copy/paste or
swapped-argument
+ * mistakes that are easy to make across the near-identical client
+ * definitions, where the holder constructor takes only strings.
+ */
+ @Test
+ public void testProductionThreadPoolConfigKeysWellFormed() {
+ String[][] clients = {
+ {AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE},
+ {AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_KEEPALIVE},
+ {AWS_STS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_KEEPALIVE},
+ {AWS_KMS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_KEEPALIVE},
+ };
+
+ List<String> allKeys = new ArrayList<>();
+ for (String[] client : clients) {
+ String enabled = client[0];
+ String size = client[1];
+ String keepAlive = client[2];
+
+ Assertions.assertThat(enabled)
+ .as("Enabled key should end with .threadpool.enabled")
+ .endsWith(".threadpool.enabled");
+ String base =
+ enabled.substring(0, enabled.length() - ".enabled".length());
+ Assertions.assertThat(size)
+ .as("Size key should be base + .size for base %s", base)
+ .isEqualTo(base + ".size");
+ Assertions.assertThat(keepAlive)
+ .as("Keepalive key should be base + .keepalive.seconds for base %s",
+ base)
+ .isEqualTo(base + ".keepalive.seconds");
+
+ allKeys.add(enabled);
+ allKeys.add(size);
+ allKeys.add(keepAlive);
+ }
+
+ Assertions.assertThat(allKeys)
+ .as("All shared thread pool config keys should be unique")
+ .doesNotHaveDuplicates();
+ }
+
+ /**
+ * Verify the shared-thread-pool defaults declared in core-default.xml match
+ * the constants used by the lazy holders: every pool is disabled by default
+ * and the size/keepalive defaults are the documented values. This catches a
+ * typo in a core-default.xml name or a mismatch between Constants, the
holder
+ * defaults and core-default.xml.
+ */
+ @Test
+ public void testProductionThreadPoolDefaultsInCoreDefaultXml() {
+ Configuration conf = new Configuration();
+ String[] enabledKeys = {
+ AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ };
+ String[] sizeKeys = {
+ AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_SIZE,
+ };
+ String[] keepAliveKeys = {
+ AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_KEEPALIVE,
+ };
+
+ for (String key : enabledKeys) {
+ Assertions.assertThat(conf.getBoolean(key, true))
+ .as("%s should default to false in core-default.xml", key)
+ .isFalse();
+ }
+ for (String key : sizeKeys) {
+ Assertions.assertThat(conf.getInt(key, -1))
+ .as("%s should match the documented default in core-default.xml",
key)
+ .isEqualTo(AWS_CLIENT_SHARED_THREADPOOL_SIZE_DEFAULT);
+ }
+ for (String key : keepAliveKeys) {
+ Assertions.assertThat(conf.getInt(key, -1))
+ .as("%s should match the documented default in core-default.xml",
key)
+ .isEqualTo(AWS_CLIENT_SHARED_THREADPOOL_KEEPALIVE_DEFAULT);
+ }
+ }
+
+ /**
+ * Verify each AWS client factory wires its shared-thread-pool holder with
the
+ * config keys for that client, in the correct slots. The holder constructor
+ * takes only strings, so a swapped or mis-pasted key would otherwise be
+ * silent; this asserts the enabled/size/keepalive keys land where intended.
+ */
+ @Test
+ public void testFactoryExecutorHoldersWiredWithExpectedKeys() {
+ assertHolderKeys(DefaultS3ClientFactory.s3SyncExecutorHolder(),
+ AWS_S3_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_CLIENT_SHARED_THREADPOOL_KEEPALIVE);
+ assertHolderKeys(DefaultS3ClientFactory.s3AsyncExecutorHolder(),
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_S3_ASYNC_CLIENT_SHARED_THREADPOOL_KEEPALIVE);
+ assertHolderKeys(STSClientFactory.stsExecutorHolder(),
+ AWS_STS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_STS_CLIENT_SHARED_THREADPOOL_KEEPALIVE);
+ assertHolderKeys(EncryptionS3ClientFactory.kmsExecutorHolder(),
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_ENABLED,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_SIZE,
+ AWS_KMS_CLIENT_SHARED_THREADPOOL_KEEPALIVE);
+ }
+
+ /**
+ * Assert a holder is wired with the expected enabled/size/keepalive keys.
+ * @param holder the holder under test
+ * @param enabledKey expected enabled key
+ * @param sizeKey expected size key
+ * @param keepAliveKey expected keepalive key
+ */
+ private static void assertHolderKeys(LazySharedThreadPoolHolder holder,
+ String enabledKey, String sizeKey, String keepAliveKey) {
+ Assertions.assertThat(holder.getEnabledKey())
+ .as("enabled key")
+ .isEqualTo(enabledKey);
+ Assertions.assertThat(holder.getSizeKey())
+ .as("size key")
+ .isEqualTo(sizeKey);
+ Assertions.assertThat(holder.getKeepAliveKey())
+ .as("keepalive key")
+ .isEqualTo(keepAliveKey);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
