Default async future completion executor is effectively single-threaded

[kevinkarpenske]

The default async future completion executor is effectively single-thread with the current construction.

    private Executor resolveAsyncFutureCompletionExecutor(SdkClientConfiguration config) {
        Supplier<Executor> defaultExecutor = () ->
                new ThreadPoolExecutor(0, 50,
                                       10, TimeUnit.SECONDS,
                                       new LinkedBlockingQueue<>(10_000),
                                       new ThreadFactoryBuilder().threadNamePrefix("sdk-async-response").build());

        return Optional.ofNullable(config.option(FUTURE_COMPLETION_EXECUTOR))
                       .orElseGet(defaultExecutor);
    }

Expected Behavior

The intent appears to be an executor that grows from 0 to 50 workers, and then queues 10,000 additional requests. However, that is not the behavior of ThreadPoolExecutor with the given configuration.

With the configuration above, the executor is limited to its core pool size (0), after which it must fill its queue (capacity 10,000), and only when the queue is full does the executor grow to the max pool size (50).

The underlying implementation in ThreadPoolExecutor ensures there is at least one worker after submission to the queue, even if core pool size is zero, hence the single-threaded behavior (as opposed to complete inoperability).

Current Behavior

Consider the following unit test:

@Test
public void testExecutor() {
    final ThreadPoolExecutor executor = new ThreadPoolExecutor(0, 50,
            10, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10_000),
            new ThreadFactoryBuilder().threadNamePrefix("sdk-async-response").build());
    final AtomicLong concurrency = new AtomicLong(0);
    final long start = System.currentTimeMillis();
    IntStream.range(0, 5)
            .mapToObj(i -> executor.submit(() -> {
                System.out.println("in: " + concurrency.incrementAndGet());
                Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
                System.out.println("out: " + concurrency.decrementAndGet());
            }))
            .collect(Collectors.toList())
            .forEach(Futures::getUnchecked);
    System.out.println("elapsed: " + (System.currentTimeMillis() - start));
}

Output:

in: 1
out: 0
in: 1
out: 0
in: 1
out: 0
in: 1
out: 0
in: 1
out: 0
elapsed: 5016

Now consider this alternative construction, which seems to reflect the original intent:

@Test
public void testExecutor() {
    final ThreadPoolExecutor executor = new ThreadPoolExecutor(50, 50,
            10, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10_000),
            new ThreadFactoryBuilder().threadNamePrefix("sdk-async-response").build());
    executor.allowCoreThreadTimeOut(true); //*important*
    final AtomicLong concurrency = new AtomicLong(0);
    final long start = System.currentTimeMillis();
    IntStream.range(0, 5)
            .mapToObj(i -> executor.submit(() -> {
                System.out.println("in: " + concurrency.incrementAndGet());
                Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
                System.out.println("out: " + concurrency.decrementAndGet());
            }))
            .collect(Collectors.toList())
            .forEach(Futures::getUnchecked);
    System.out.println("elapsed: " + (System.currentTimeMillis() - start));
}

Output:

in: 1
in: 2
in: 3
in: 4
in: 5
out: 4
out: 2
out: 1
out: 3
out: 0
elapsed: 1006

Context

Identified while debugging an application and noticing that requests using thenApply or thenCompose on a CompletableFuture returned by the SDK were being funneled through a single thread in the sdk-async-response thread pool.

Your Environment

• AWS Java SDK version used: 2.1.1
• JDK version used: 1.8.0_131

[zoewangg]

Thank you so much for reporting this! We will work on the fix.

[zoewangg]

The fix has been released in 2.3.0. Closing the issue.