[
https://issues.apache.org/jira/browse/HADOOP-19862?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18089440#comment-18089440
]
ASF GitHub Bot commented on HADOOP-19862:
-----------------------------------------
ajfabbri commented on code in PR #8426:
URL: https://github.com/apache/hadoop/pull/8426#discussion_r3422764384
##########
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);
+ checkArgument(keepAlive > 0,
+ "Value of %s must be positive, got: %s", keepAliveKey, keepAlive);
+ 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
+ */
Review Comment:
Should we add `@VisibleForTesting` annotation here? Assuming that is the
intent.
##########
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;
Review Comment:
`volatile` is redundant here since the field is only accessed under `this`
object's monitor lock (i.e. `synchronized` method).
##########
hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestSharedScheduledExecutor.java:
##########
@@ -0,0 +1,280 @@
+/*
+ * 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.impl.LazySharedThreadPoolHolder;
+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 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;
+
+/**
+ * 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.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("must be positive");
+ }
+
+ @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("must be positive");
+ }
+
+ @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("must be positive");
+ }
+
+ @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("must be positive");
+ }
+
+ @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();
+ executor.shutdown();
Review Comment:
Wouldn't hurt to assert that the values (size, keepalive) are correct
(sepecially since the constructor takes all strings, so it is easy to misplace
/ swap arguments).
##########
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";
Review Comment:
Can you please add corresponding entries to
`hadoop-common-project/hadoop-common/src/main/resources/core-default.xml`,
populated with default values?
> 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]
