[
https://issues.apache.org/jira/browse/HADOOP-19862?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18089441#comment-18089441
]
ASF GitHub Bot commented on HADOOP-19862:
-----------------------------------------
ajfabbri commented on code in PR #8426:
URL: https://github.com/apache/hadoop/pull/8426#discussion_r3422798643
##########
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). (This comment is a nit: totally optional to address or not)
> 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]
