Update the pool size for asyncFutureCompletionPool and validate peak thread count in stability tests

Author: zoewangg

.changes/next-release/bugfix-AWSSDKforJavav2-045e79c.json

@@ -0,0 +1,5 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "type": "bugfix", 
+    "description": "Update the pool size for default async future completion executor service. See [#1251](https://github.com/aws/aws-sdk-java-v2/issues/1251), [#994](https://github.com/aws/aws-sdk-java-v2/issues/994)"
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java

@@ -264,16 +264,20 @@ protected AttributeMap childHttpConfig() {
     }
 
     /**
-     * Finalize which async executor service will be used for the created client.
+     * Finalize which async executor service will be used for the created client. The default async executor
+     * service has at least 8 core threads and can scale up to at least 64 threads when needed depending
+     * on the number of processors available.
      */
     private Executor resolveAsyncFutureCompletionExecutor(SdkClientConfiguration config) {
         Supplier<Executor> defaultExecutor = () -> {
-            ThreadPoolExecutor executor = new ThreadPoolExecutor(50, 50,
+            int processors = Runtime.getRuntime().availableProcessors();
+            int corePoolSize = Math.max(8, processors);
+            int maxPoolSize = Math.max(64, processors * 2);
+            ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
                                                                  10, TimeUnit.SECONDS,
-                                                                 new LinkedBlockingQueue<>(10_000),
+                                                                 new LinkedBlockingQueue<>(1_000),
                                                                  new ThreadFactoryBuilder()
                                                                      .threadNamePrefix("sdk-async-response").build());
-
             // Allow idle core threads to time out
             executor.allowCoreThreadTimeOut(true);
             return executor;
@@ -421,4 +425,5 @@ public void close() {
         }
     }
 
+
 }

test/stability-tests/src/it/java/software/amazon/awssdk/stability/tests/TestResult.java

@@ -26,6 +26,8 @@ public class TestResult {
     private final int ioExceptionCount;
     private final int clientExceptionCount;
     private final int unknownExceptionCount;
+    private final int peakThreadCount;
+    private final double heapMemoryAfterGCUsage;
 
     private TestResult(Builder builder) {
         this.testName = builder.testName;
@@ -34,6 +36,8 @@ private TestResult(Builder builder) {
         this.ioExceptionCount = builder.ioExceptionCount;
         this.clientExceptionCount = builder.clientExceptionCount;
         this.unknownExceptionCount = builder.unknownExceptionCount;
+        this.heapMemoryAfterGCUsage = builder.heapMemoryAfterGCUsage;
+        this.peakThreadCount = builder.peakThreadCount;
     }
 
     public static Builder builder() {
@@ -64,6 +68,14 @@ public int unknownExceptionCount() {
         return unknownExceptionCount;
     }
 
+    public int peakThreadCount() {
+        return peakThreadCount;
+    }
+
+    public double heapMemoryAfterGCUsage() {
+        return heapMemoryAfterGCUsage;
+    }
+
     @Override
     public String toString() {
         return "{" +
@@ -73,6 +85,8 @@ public String toString() {
                ", ioExceptionCount: " + ioExceptionCount +
                ", clientExceptionCount: " + clientExceptionCount +
                ", unknownExceptionCount: " + unknownExceptionCount +
+               ", peakThreadCount: " + peakThreadCount +
+               ", heapMemoryAfterGCUsage: " + heapMemoryAfterGCUsage +
                '}';
     }
 
@@ -83,6 +97,8 @@ public static class Builder {
         private int ioExceptionCount;
         private int clientExceptionCount;
         private int unknownExceptionCount;
+        private int peakThreadCount;
+        private double heapMemoryAfterGCUsage;
 
         private Builder() {
         }
@@ -117,6 +133,16 @@ public Builder unknownExceptionCount(int unknownExceptionCount) {
             return this;
         }
 
+        public Builder peakThreadCount(int peakThreadCount) {
+            this.peakThreadCount = peakThreadCount;
+            return this;
+        }
+
+        public Builder heapMemoryAfterGCUsage(double heapMemoryAfterGCUsage) {
+            this.heapMemoryAfterGCUsage = heapMemoryAfterGCUsage;
+            return this;
+        }
+
         public TestResult build() {
             return new TestResult(this);
         }

test/stability-tests/src/it/java/software/amazon/awssdk/stability/tests/s3/S3AsyncStabilityTest.java

@@ -45,6 +45,7 @@ public static void setup() {
     @AfterAll
     public static void cleanup() {
         deleteBucketAndAllContents(bucketName);
+        s3NettyClient.close();
     }
 
     @RetryableTest(maxRetries = 3, retryableException = StabilityTestsRetryableException.class)

test/stability-tests/src/it/java/software/amazon/awssdk/stability/tests/sqs/SqsAsyncStabilityTest.java

@@ -47,6 +47,7 @@ public static void tearDown() {
         if (queueUrl != null) {
             sqsAsyncClient.deleteQueue(b -> b.queueUrl(queueUrl));
         }
+        sqsAsyncClient.close();
     }
 
     @RetryableTest(maxRetries = 3, retryableException = StabilityTestsRetryableException.class)

test/stability-tests/src/it/java/software/amazon/awssdk/stability/tests/transcribestreaming/TranscribeStreamingStabilityTest.java

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.IntFunction;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -63,6 +64,11 @@ public static void setup() {
         }
     }
 
+    @AfterAll
+    public static void tearDown() {
+        transcribeStreamingClient.close();
+    }
+
     @RetryableTest(maxRetries = 3, retryableException = StabilityTestsRetryableException.class)
     public void startTranscription() {
         IntFunction<CompletableFuture<?>> futureIntFunction = i ->

test/stability-tests/src/it/java/software/amazon/awssdk/stability/tests/utils/RetryableTestExtension.java

@@ -104,7 +104,7 @@ private static final class RetryableTestsTemplateInvocationContext implements Te
     private static final class RetryExecutor implements Iterator<RetryableTestsTemplateInvocationContext> {
         private final int maxRetries;
         private int totalAttempts;
-        private static final List<Throwable> throwables = new ArrayList<>();
+        private List<Throwable> throwables = new ArrayList<>();
 
         RetryExecutor(int maxRetries) {
             this.maxRetries = maxRetries;

test/stability-tests/src/it/java/software/amazon/awssdk/stability/tests/utils/StabilityTestRunner.java

@@ -16,7 +16,13 @@
 package software.amazon.awssdk.stability.tests.utils;
 
 
+import static java.lang.management.MemoryType.HEAP;
+
 import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryPoolMXBean;
+import java.lang.management.MemoryUsage;
+import java.lang.management.ThreadMXBean;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
@@ -65,15 +71,23 @@ public class StabilityTestRunner {
     private static final Logger log = Logger.loggerFor(StabilityTestRunner.class);
     private static final double ALLOWED_FAILURE_RATIO = 0.05;
     private static final int TESTS_TIMEOUT_IN_MINUTES = 60;
+    // The peak thread count might be different depending on the machine the tests are currently running on.
+    // because of the internal thread pool used in AsynchronousFileChannel
+    private static final int ALLOWED_PEAK_THREAD_COUNT = 60;
 
+    private ThreadMXBean threadMXBean;
     private IntFunction<CompletableFuture<?>> futureFactory;
     private List<CompletableFuture<?>> futures;
     private String testName;
     private Duration delay = Duration.ZERO;
     private Integer requestCountPerRun;
     private Integer totalRuns = 1;
 
+
     private StabilityTestRunner() {
+        threadMXBean = ManagementFactory.getThreadMXBean();
+        // Reset peak thread count for every test
+        threadMXBean.resetPeakThreadCount();
     }
 
     /**
@@ -177,6 +191,26 @@ private TestResult runTestsFromFutureFunction() {
         return generateTestResult(totalRequestNumber, testName, exceptionCounter, completableFutures);
     }
 
+    private double calculateHeapMemoryAfterGCUsage() {
+        List<MemoryPoolMXBean> memoryPoolMXBeans = ManagementFactory.getMemoryPoolMXBeans();
+
+        long used = 0, max = 0;
+
+        for (MemoryPoolMXBean memoryPoolMXBean : memoryPoolMXBeans) {
+            String name = memoryPoolMXBean.getName();
+
+            if (!name.contains("Eden")) {
+                if (memoryPoolMXBean.getType().equals(HEAP)) {
+                    MemoryUsage memoryUsage = memoryPoolMXBean.getCollectionUsage();
+                    used += memoryUsage.getUsed();
+                    max += memoryUsage.getMax() == -1 ? 0 : memoryUsage.getMax();
+                }
+            }
+        }
+
+        return used / (double) max;
+    }
+
     private TestResult runTestsFromFutures() {
         ExceptionCounter exceptionCounter = new ExceptionCounter();
         CompletableFuture[] completableFutures =
@@ -229,7 +263,7 @@ private static boolean isIOException(Throwable throwable) {
         return throwable.getClass().isAssignableFrom(IOException.class);
     }
 
-    private static TestResult generateTestResult(int totalRequestNumber, String testName, ExceptionCounter exceptionCounter,
+    private TestResult generateTestResult(int totalRequestNumber, String testName, ExceptionCounter exceptionCounter,
                                                  CompletableFuture[] completableFutures) {
         try {
             CompletableFuture.allOf(completableFutures).get(TESTS_TIMEOUT_IN_MINUTES, TimeUnit.MINUTES);
@@ -247,6 +281,8 @@ private static TestResult generateTestResult(int totalRequestNumber, String test
                          .ioExceptionCount(exceptionCounter.ioExceptionCount())
                          .totalRequestCount(totalRequestNumber)
                          .unknownExceptionCount(exceptionCounter.unknownExceptionCount())
+                         .peakThreadCount(threadMXBean.getPeakThreadCount())
+                         .heapMemoryAfterGCUsage(calculateHeapMemoryAfterGCUsage())
                          .build();
     }
 
@@ -286,6 +322,11 @@ private static void processResult(TestResult testResult) {
                                           ALLOWED_FAILURE_RATIO * 100, ratio * 100);
             throw new StabilityTestsRetryableException(errorMessage);
         }
-    }
 
+        if (testResult.peakThreadCount() > ALLOWED_PEAK_THREAD_COUNT) {
+            String errorMessage = String.format("The number of peak thread exceeds the allowed peakThread threshold %s",
+                                                ALLOWED_PEAK_THREAD_COUNT);
+            throw new AssertionError(errorMessage);
+        }
+    }
 }