Resource Leak When Using S3Client.getObject() with ResponseTransformer.toInputStream()

[dev-k13]

Describe the bug

A resource leak is observed when retrieving an Amazon S3 object as a stream using S3Client.getObject() with ResponseTransformer.toInputStream(). This leak occurs despite using both try-with-resources for automatic resource management and manual calls to close() on the InputStream.

Thread dump observation:
The thread response-input-stream-timeout-scheduler is visible in the dump.

Regression Issue

• Select this option if this issue appears to be a regression.

Expected Behavior

When the ResponseInputStream returned by s3Client.getObject(..., ResponseTransformer.toInputStream()) is closed, all associated resources should be released and closed.

Current Behavior

The thread response-input-stream-timeout-scheduler is visible in the dump preventing the module to perform a clean and graceful shutdown.

Reproduction Steps

Code Snippet:

import org.apache.commons.io.IOUtils;

try (InputStream inputStream = s3Client.getObject(getObjectRequest, ResponseTransformer.toInputStream())) {
    return IOUtils.toString(inputStream, StandardCharsets.UTF_8);
}

Possible Solution

No response

Additional Information/Context

As a workaround, s3Client.getObject(getObjectRequest, ResponseTransformer.toBytes()).asByteArray(); was used to eliminate stream management.

AWS Java SDK version used

2.33.4

JDK version used

JDK17

Operating System and version

Linux

[bhoradc]

Hi @dev-k13

Thanks for reporting the issue. In SDK v2.32.32 we added a timeout functionality to ResponseInputStream.

It does appear from this version onwards SDK is creating a persistent background thread - response-input-stream-timeout-scheduler when using ResponseTransformer.toInputStream() that never gets cleaned up, even after proper stream closure and S3Client shutdown. Reference - TimeoutScheduler

As a workaround, ResponseTransformer.toBytes() would work or ResponseTransformer.toInputStream(Duration.ZERO) to disable the timeout feature to avoid the leak.

ReproSample

package org.example;

import org.apache.commons.io.IOUtils;
import software.amazon.awssdk.core.sync.ResponseTransformer;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.Duration;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        S3Client s3Client = S3Client.builder().region(Region.US_EAST_1).build();
        
        GetObjectRequest request = GetObjectRequest.builder()
                .bucket("***-test-bucket")
                .key("test-key")
                .build();

        try (InputStream inputStream = s3Client.getObject(request, ResponseTransformer.toInputStream())) {
            String content = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
            System.out.println("Content: " + content);
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        } finally {
            s3Client.close();
        }

        
        System.out.println("\nLooking for response-input-stream-timeout-scheduler thread:");
        Thread.getAllStackTraces().forEach((thread, stackTrace) -> {
            if (thread.getName().contains("response-input-stream-timeout-scheduler")) {
                System.out.println("\"" + thread.getName() + "\" #" + thread.getId() + 
                        " daemon=" + thread.isDaemon() + " prio=" + thread.getPriority() + 
                        " state=" + thread.getState());
                for (StackTraceElement element : stackTrace) {
                    System.out.println("\tat " + element);
                }
            }
        });
        Thread.sleep(20000);
    }
}

Result

Looking for response-input-stream-timeout-scheduler thread:
"response-input-stream-timeout-scheduler" #38 daemon=true prio=5 state=WAITING
	at java.base@22.0.2/jdk.internal.misc.Unsafe.park(Native Method)
	at java.base@22.0.2/java.util.concurrent.locks.LockSupport.park(LockSupport.java:371)
	at java.base@22.0.2/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionNode.block(AbstractQueuedSynchronizer.java:519)
	at java.base@22.0.2/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:4011)
	at java.base@22.0.2/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3959)
	at java.base@22.0.2/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1707)
	at java.base@22.0.2/java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1170)
	at java.base@22.0.2/java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:899)
	at java.base@22.0.2/java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1070)
	at java.base@22.0.2/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
	at java.base@22.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base@22.0.2/java.lang.Thread.runWith(Thread.java:1583)
	at java.base@22.0.2/java.lang.Thread.run(Thread.java:1570)

Process finished with exit code 130 (interrupted by signal 2:SIGINT)

We will investigate this further.

Regards,
Chaitanya

[bhoradc]

Hi @dev-k13

While we work on the fix for this issue, can you kindly share how you encountered the resource leak? Ensuring that we also understand any other larger potential impact, if there's any.

Regards,
Chaitanya

[dev-k13]

Hi @dev-k13

Thanks for reporting the issue. In SDK v2.32.32 we added a timeout functionality to ResponseInputStream.

It does appear from this version onwards SDK is creating a persistent background thread - response-input-stream-timeout-scheduler when using ResponseTransformer.toInputStream() that never gets cleaned up, even after proper stream closure and S3Client shutdown. Reference - TimeoutScheduler

As a workaround, ResponseTransformer.toBytes() would work or ResponseTransformer.toInputStream(Duration.ZERO) to disable the timeout feature to avoid the leak.

ReproSample
package org.example;

import org.apache.commons.io.IOUtils;
import software.amazon.awssdk.core.sync.ResponseTransformer;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.Duration;

public class Main {
public static void main(String[] args) throws InterruptedException {
S3Client s3Client = S3Client.builder().region(Region.US_EAST_1).build();

    GetObjectRequest request = GetObjectRequest.builder()
            .bucket("***-test-bucket")
            .key("test-key")
            .build();

    try (InputStream inputStream = s3Client.getObject(request, ResponseTransformer.toInputStream())) {
        String content = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
        System.out.println("Content: " + content);
    } catch (Exception e) {
        System.err.println("Error: " + e.getMessage());
    } finally {
        s3Client.close();
    }

    
    System.out.println("\nLooking for response-input-stream-timeout-scheduler thread:");
    Thread.getAllStackTraces().forEach((thread, stackTrace) -> {
        if (thread.getName().contains("response-input-stream-timeout-scheduler")) {
            System.out.println("\"" + thread.getName() + "\" #" + thread.getId() + 
                    " daemon=" + thread.isDaemon() + " prio=" + thread.getPriority() + 
                    " state=" + thread.getState());
            for (StackTraceElement element : stackTrace) {
                System.out.println("\tat " + element);
            }
        }
    });
    Thread.sleep(20000);
}

}
Result

Looking for response-input-stream-timeout-scheduler thread:
"response-input-stream-timeout-scheduler" #38 daemon=true prio=5 state=WAITING
	at java.base@22.0.2/jdk.internal.misc.Unsafe.park(Native Method)
	at java.base@22.0.2/java.util.concurrent.locks.LockSupport.park(LockSupport.java:371)
	at java.base@22.0.2/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionNode.block(AbstractQueuedSynchronizer.java:519)
	at java.base@22.0.2/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:4011)
	at java.base@22.0.2/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3959)
	at java.base@22.0.2/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1707)
	at java.base@22.0.2/java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1170)
	at java.base@22.0.2/java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:899)
	at java.base@22.0.2/java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1070)
	at java.base@22.0.2/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
	at java.base@22.0.2/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base@22.0.2/java.lang.Thread.runWith(Thread.java:1583)
	at java.base@22.0.2/java.lang.Thread.run(Thread.java:1570)

Process finished with exit code 130 (interrupted by signal 2:SIGINT)

We will investigate this further.

Regards, Chaitanya

Appreciate the prompt response. The suggested workaround ResponseTransformer.toInputStream(Duration.ZERO) to disable the timeout feature works; we will go with this approach for now. Thank you.

[dev-k13]

Hi @dev-k13

While we work on the fix for this issue, can you kindly share how you encountered the resource leak? Ensuring that we also understand any other larger potential impact, if there's any.

Regards, Chaitanya

Hi @bhoradc,

We encountered it during application (JVM) shutdown. The leak prevents the application from performing a clean/ graceful shutdown but did not impact anything (AFAIK) during runtime.

[marchof]

Here is another scenario where I encountered this thread "leak". When running S3 client code within a Maven build with exec:java the following warnings show up the the build log:

Warning:  thread Thread[#82,response-input-stream-timeout-scheduler,5,io.javaalmanac.find.indexer.Indexer] was interrupted but is still alive after waiting at least 15000msecs
Warning:  thread Thread[#82,response-input-stream-timeout-scheduler,5,io.javaalmanac.find.indexer.Indexer] will linger despite being asked to die via interruption
Warning:  NOTE: 1 thread(s) did not finish despite being asked to via interruption. This is not a problem with exec:java, it is a problem with the running code. Although not serious, it should be remedied.

Adding ResponseTransformer.toInputStream(Duration.ZERO) fixes the warning.