[
https://issues.apache.org/jira/browse/HADOOP-19862?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090631#comment-18090631
]
ASF GitHub Bot commented on HADOOP-19862:
-----------------------------------------
steveloughran commented on code in PR #8426:
URL: https://github.com/apache/hadoop/pull/8426#discussion_r3087252644
##########
hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/LazySharedThreadPoolHolder.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.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.util.Preconditions.checkArgument;
+
+/**
+ * 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 volatile 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 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 = conf.getInt(sizeKey,
AWS_CLIENT_SHARED_THREADPOOL_SIZE_DEFAULT);
+ int keepAlive = conf.getInt(keepAliveKey,
AWS_CLIENT_SHARED_THREADPOOL_KEEPALIVE_DEFAULT);
+ checkArgument(poolSize > 0,
+ "Value of %s must be positive, got: %s", sizeKey, poolSize);
Review Comment:
S3AUtils.intOption() can combine range checks with retrieval
##########
hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolEnabled.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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 stay isolated because hadoop-aws runs
+ * tests with reuseForks=false (a fresh JVM per test class); 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 {
Review Comment:
does test class teardown need to do anything about the pools?
##########
hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ASharedThreadPoolEnabled.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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 stay isolated because hadoop-aws runs
+ * tests with reuseForks=false (a fresh JVM per test class); 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();
+ // 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);
Review Comment:
use removeBaseAndBucketOverrides() to remove bucket overrides fo these
before setting them; really screws up test results if you ever actually set
them on a bucket...
> S3A: Thread leak from AWS SDK v2 ScheduledExecutorService
> ---------------------------------------------------------
>
> Key: HADOOP-19862
> URL: https://issues.apache.org/jira/browse/HADOOP-19862
> Project: Hadoop Common
> Issue Type: Bug
> Reporter: Konstantin Bereznyakov
> Priority: Major
> Labels: pull-request-available
>
> AWS SDK v2 S3 clients create internal ScheduledExecutorService instances that
> accumulate over time, causing unbounded thread growth. Thread dumps show
> thousands of sdk-ScheduledExecutor-* threads in processes that create
> multiple S3AFileSystem instances.
> Environment
> - Hadoop 3.4.x with AWS SDK v2
> - Not observed in Hadoop 3.3.x (AWS SDK v1)
> Observed Behavior
> Thread dump comparison:
> Hadoop 3.3.x (AWS SDK v1): Normal thread count
> Hadoop 3.4.x (AWS SDK v2): 1600+ "sdk-ScheduledExecutor-*" threads
> Thread pattern:
> "sdk-ScheduledExecutor-0-0" daemon prio=5 waiting
> "sdk-ScheduledExecutor-0-1" daemon prio=5 waiting
> ...
> "sdk-ScheduledExecutor-0-4" daemon prio=5 waiting
> "sdk-ScheduledExecutor-1-0" daemon prio=5 waiting
> ...
> Root Cause
> AWS SDK v2's SdkDefaultClientBuilder creates a ScheduledThreadPoolExecutor
> with 5 threads per client when no executor is explicitly provided
> (https://github.com/aws/aws-sdk-java-v2/issues/1690):
> Executors.newScheduledThreadPool(5,
> new
> ThreadFactoryBuilder().threadNamePrefix("sdk-ScheduledExecutor").build())
> These threads are used for retry scheduling, timeout handling, and
> credential refresh.
> Contributing Factors
> 1. AbstractFileSystem has no caching
> Unlike FileSystem.get() which uses CACHE.get(uri, conf),
> AbstractFileSystem.get() always creates new instances:
> // AbstractFileSystem.java:263-266
> public static AbstractFileSystem get(final URI uri, final Configuration
> conf) {
> return createFileSystem(uri, conf); // NO CACHING
> }
> Each FileContext.getFileContext() call with an S3 URI creates:
> - New AbstractFileSystem (S3A)
> - New S3AFileSystem
> - New S3Client
> - 5 new sdk-ScheduledExecutor threads
> 2. S3Client threads not released on close
> As documented in https://github.com/aws/aws-sdk-java-v2/issues/1690:
> "When using cached, ephemeral clients, I can see that the scheduled thread
> pool will at times be leaked when the aws client is evicted"
> 3. Multiple client types affected
> S3A creates multiple AWS SDK clients:
> - S3Client (sync)
> - S3AsyncClient
> - STS client (for delegation tokens)
> - KMS client (for encryption)
> Each client instance creates its own 5-thread pool.
> Impact
> - Unbounded thread growth in any process using S3A
> - Resource exhaustion leading to OOM or system instability
> - Particularly affects YARN NodeManager, Spark drivers/executors, and other
> services that create many filesystem instances
> Related
> - https://github.com/aws/aws-sdk-java-v2/issues/1690 - SDK issue
> documenting the problem
> - https://github.com/aws/aws-sdk-java-v2/pull/4002 - SDK fix allowing
> shared executor configuration
> - HADOOP-19624 - Similar thread leak in ABFS (AbfsClientThrottlingAnalyzer)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
