Skip to content

Add keepalive to responseInputStream timeout scheduler executor to ma… #6756

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RanVaknin merged 7 commits into from
Mar 4, 2026
+26 −6
Merged

Add keepalive to responseInputStream timeout scheduler executor to ma… #6756

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
aeeef69
Add keepalive to responseInputStream timeout scheduler executor to ma…
RanVaknin Feb 27, 2026
a0d4a11
Fix checkstyle
RanVaknin Mar 2, 2026
9a1ce25
Add changelog
RanVaknin Mar 2, 2026
463c47b
Separate timeout value, and extend fix to ResponsePublisher
RanVaknin Mar 2, 2026
49be954
Fix changelog
RanVaknin Mar 2, 2026
4310e47
Update changelog
RanVaknin Mar 3, 2026
842b12b
Fix changelog
RanVaknin Mar 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .changes/next-release/bugfix-AWSSDKforJavav2-7d9ef28.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
{
"type": "bugfix",
"category": "AWS SDK for Java v2",
"contributor": "",
"description": "Fixed a thread leak in ResponseInputStream and ResponsePublisher where the internal timeout scheduler thread persisted for the lifetime of the JVM, even when no streams were active. The thread now terminates after being idle for 60 seconds."
}
13 changes: 10 additions & 3 deletions core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,9 @@
import java.io.IOException;
import java.io.InputStream;
import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import software.amazon.awssdk.annotations.SdkPublicApi;
import software.amazon.awssdk.annotations.SdkTestInternalApi;
Expand Down Expand Up @@ -61,6 +61,7 @@ public final class ResponseInputStream<ResponseT> extends SdkFilterInputStream i

private static final Logger log = Logger.loggerFor(ResponseInputStream.class);
private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(60);
private static final long THREAD_IDLE_TIMEOUT_SECONDS = 60;
private final ResponseT response;
private final Abortable abortable;
private ScheduledFuture<?> timeoutTask;
Expand Down Expand Up @@ -140,12 +141,18 @@ private void scheduleTimeoutTask(Duration timeout) {
}

private static final class TimeoutScheduler {
static final ScheduledExecutorService INSTANCE =
Executors.newScheduledThreadPool(1, r -> {
static final ScheduledExecutorService INSTANCE;

static {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, r -> {
Thread t = new Thread(r, "response-input-stream-timeout-scheduler");
t.setDaemon(true);
return t;
});
executor.setKeepAliveTime(THREAD_IDLE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
executor.allowCoreThreadTimeOut(true);
INSTANCE = executor;
}
}

/**
Expand Down
13 changes: 10 additions & 3 deletions core/sdk-core/src/main/java/software/amazon/awssdk/core/async/ResponsePublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,9 @@
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
Expand Down Expand Up @@ -50,6 +50,7 @@ public final class ResponsePublisher<ResponseT extends SdkResponse> implements S

private static final Logger log = Logger.loggerFor(ResponsePublisher.class);
private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(60);
private static final long THREAD_IDLE_TIMEOUT_SECONDS = 60;
private final ResponseT response;
private final SdkPublisher<ByteBuffer> publisher;
private ScheduledFuture<?> timeoutTask;
Expand Down Expand Up @@ -101,12 +102,18 @@ private void scheduleTimeoutTask(Duration timeout) {
}

private static final class TimeoutScheduler {
static final ScheduledExecutorService INSTANCE =
Executors.newScheduledThreadPool(1, r -> {
static final ScheduledExecutorService INSTANCE;

static {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, r -> {
Thread t = new Thread(r, "response-publisher-timeout-scheduler");
t.setDaemon(true);
return t;
});
executor.setKeepAliveTime(THREAD_IDLE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
executor.allowCoreThreadTimeOut(true);
INSTANCE = executor;
}
}

private static class CancellingSubscriber implements Subscriber<ByteBuffer> {
Expand Down
Loading