From 073d2eb56dad74bdf1d8f524d93c3977beac7907 Mon Sep 17 00:00:00 2001 From: Alexander Farber Date: Wed, 9 Sep 2026 16:52:11 +0200 Subject: [PATCH] Close HTTP response streams when interceptors throw --- .../bugfix-AWSSDKforJavav2-4927ecb.json | 6 + .../ExecutionInterceptorChain.java | 38 +++-- ...ransmissionExecutionInterceptorsStage.java | 14 +- ...marshallingExecutionInterceptorsStage.java | 15 +- ...ClientHandlerInterceptorExceptionTest.java | 159 ++++++++++++++++++ 5 files changed, 216 insertions(+), 16 deletions(-) create mode 100644 .changes/next-release/bugfix-AWSSDKforJavav2-4927ecb.json create mode 100644 core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/SyncClientHandlerInterceptorExceptionTest.java diff --git a/.changes/next-release/bugfix-AWSSDKforJavav2-4927ecb.json b/.changes/next-release/bugfix-AWSSDKforJavav2-4927ecb.json new file mode 100644 index 000000000000..a5c526b0b009 --- /dev/null +++ b/.changes/next-release/bugfix-AWSSDKforJavav2-4927ecb.json @@ -0,0 +1,6 @@ +{ + "type": "bugfix", + "category": "AWS SDK for Java v2", + "contributor": "afarber", + "description": "Close synchronous HTTP response streams when response interceptors throw exceptions." +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/interceptor/ExecutionInterceptorChain.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/interceptor/ExecutionInterceptorChain.java index cd2bff920ffd..3a888f1b450b 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/interceptor/ExecutionInterceptorChain.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/interceptor/ExecutionInterceptorChain.java @@ -30,6 +30,7 @@ import software.amazon.awssdk.core.sync.RequestBody; import software.amazon.awssdk.http.SdkHttpRequest; import software.amazon.awssdk.http.SdkHttpResponse; +import software.amazon.awssdk.utils.IoUtils; import software.amazon.awssdk.utils.Logger; import software.amazon.awssdk.utils.Validate; @@ -112,22 +113,33 @@ public void afterTransmission(Context.AfterTransmission context, ExecutionAttrib public InterceptorContext modifyHttpResponse(InterceptorContext context, ExecutionAttributes executionAttributes) { InterceptorContext result = context; - - for (int i = interceptors.size() - 1; i >= 0; i--) { - SdkHttpResponse interceptorResult = - interceptors.get(i).modifyHttpResponse(result, executionAttributes); - InputStream response = interceptors.get(i).modifyHttpResponseContent(result, executionAttributes).orElse(null); - - if (interceptorResult != result.httpResponse() || response != result.responseBody().orElse(null)) { - validateInterceptorResult(result.httpResponse(), interceptorResult, interceptors.get(i), "modifyHttpResponse"); - result = result.copy(r -> r.httpResponse(interceptorResult) - .responseBody(response)); + InputStream response = context.responseBody().orElse(null); + boolean completed = false; + + try { + for (int i = interceptors.size() - 1; i >= 0; i--) { + SdkHttpResponse interceptorResult = + interceptors.get(i).modifyHttpResponse(result, executionAttributes); + response = interceptors.get(i) + .modifyHttpResponseContent(result, executionAttributes) + .orElse(null); + + if (interceptorResult != result.httpResponse() || response != result.responseBody().orElse(null)) { + validateInterceptorResult(result.httpResponse(), interceptorResult, interceptors.get(i), + "modifyHttpResponse"); + InputStream currentResponse = response; + result = result.copy(r -> r.httpResponse(interceptorResult) + .responseBody(currentResponse)); + } } - + completed = true; + return result; + } finally { + if (!completed) { + IoUtils.closeQuietlyV2(response, LOG); + } } - - return result; } public InterceptorContext modifyAsyncHttpResponse(InterceptorContext context, diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java index a7cada02b06c..b7326237dd3b 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java @@ -23,11 +23,15 @@ import software.amazon.awssdk.http.AbortableInputStream; import software.amazon.awssdk.http.SdkHttpFullRequest; import software.amazon.awssdk.http.SdkHttpFullResponse; +import software.amazon.awssdk.utils.IoUtils; +import software.amazon.awssdk.utils.Logger; import software.amazon.awssdk.utils.Pair; @SdkInternalApi public class AfterTransmissionExecutionInterceptorsStage implements RequestPipeline, Pair> { + private static final Logger LOG = Logger.loggerFor(AfterTransmissionExecutionInterceptorsStage.class); + @Override public Pair execute(Pair input, RequestExecutionContext context) throws Exception { @@ -40,7 +44,15 @@ public Pair execute(Pair IoUtils.closeQuietlyV2(stream, LOG)); + } + } // interceptors.modifyHttpResponse interceptorContext = context.interceptorChain().modifyHttpResponse(interceptorContext, context.executionAttributes()); diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/BeforeUnmarshallingExecutionInterceptorsStage.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/BeforeUnmarshallingExecutionInterceptorsStage.java index 2e6c89a347d9..dc2fd7955eb4 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/BeforeUnmarshallingExecutionInterceptorsStage.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/BeforeUnmarshallingExecutionInterceptorsStage.java @@ -23,6 +23,8 @@ import software.amazon.awssdk.http.SdkHttpFullRequest; import software.amazon.awssdk.http.SdkHttpFullResponse; import software.amazon.awssdk.http.SdkHttpResponse; +import software.amazon.awssdk.utils.IoUtils; +import software.amazon.awssdk.utils.Logger; import software.amazon.awssdk.utils.Pair; /** @@ -32,12 +34,21 @@ @SdkInternalApi public class BeforeUnmarshallingExecutionInterceptorsStage implements RequestPipeline, SdkHttpFullResponse> { + private static final Logger LOG = Logger.loggerFor(BeforeUnmarshallingExecutionInterceptorsStage.class); @Override public SdkHttpFullResponse execute(Pair input, RequestExecutionContext context) throws Exception { - context.interceptorChain().beforeUnmarshalling(context.executionContext().interceptorContext(), - context.executionAttributes()); + boolean completed = false; + try { + context.interceptorChain().beforeUnmarshalling(context.executionContext().interceptorContext(), + context.executionAttributes()); + completed = true; + } finally { + if (!completed) { + input.right().content().ifPresent(stream -> IoUtils.closeQuietlyV2(stream, LOG)); + } + } InterruptMonitor.checkInterrupted(input.right()); return input.right(); } diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/SyncClientHandlerInterceptorExceptionTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/SyncClientHandlerInterceptorExceptionTest.java new file mode 100644 index 000000000000..cf592e57e4df --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/SyncClientHandlerInterceptorExceptionTest.java @@ -0,0 +1,159 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file. This file 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 software.amazon.awssdk.core.client.handler; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.InputStream; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Optional; +import java.util.stream.Collectors; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import software.amazon.awssdk.core.SdkRequest; +import software.amazon.awssdk.core.SdkResponse; +import software.amazon.awssdk.core.client.config.SdkClientConfiguration; +import software.amazon.awssdk.core.client.config.SdkClientOption; +import software.amazon.awssdk.core.exception.SdkServiceException; +import software.amazon.awssdk.core.http.HttpResponseHandler; +import software.amazon.awssdk.core.interceptor.Context; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.core.protocol.VoidSdkResponse; +import software.amazon.awssdk.core.runtime.transform.Marshaller; +import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.http.ExecutableHttpRequest; +import software.amazon.awssdk.http.HttpExecuteResponse; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.SdkHttpFullRequest; +import software.amazon.awssdk.http.SdkHttpResponse; +import software.amazon.awssdk.retries.DefaultRetryStrategy; +import utils.HttpTestUtils; +import utils.ValidSdkObjects; + +@RunWith(Parameterized.class) +public class SyncClientHandlerInterceptorExceptionTest { + private final SdkRequest request = mock(SdkRequest.class); + private final SdkHttpClient httpClient = mock(SdkHttpClient.class); + private final ExecutableHttpRequest httpClientCall = mock(ExecutableHttpRequest.class); + private final Marshaller marshaller = mock(Marshaller.class); + private final HttpResponseHandler responseHandler = mock(HttpResponseHandler.class); + private final HttpResponseHandler errorResponseHandler = mock(HttpResponseHandler.class); + private final InputStream responseBody = mock(InputStream.class); + + private final Hook hook; + private SdkSyncClientHandler clientHandler; + + @Parameterized.Parameters(name = "Interceptor Hook: {0}") + public static Collection data() { + return Arrays.stream(Hook.values()) + .map(hook -> new Object[] {hook}) + .collect(Collectors.toList()); + } + + public SyncClientHandlerInterceptorExceptionTest(Hook hook) { + this.hook = hook; + } + + @Before + public void setUp() throws Exception { + clientHandler = new SdkSyncClientHandler(clientConfiguration()); + + when(request.overrideConfiguration()).thenReturn(Optional.empty()); + when(marshaller.marshall(request)).thenReturn(ValidSdkObjects.sdkHttpFullRequest().build()); + when(httpClient.prepareRequest(any())).thenReturn(httpClientCall); + when(httpClientCall.call()).thenReturn(HttpExecuteResponse.builder() + .response(SdkHttpResponse.builder().statusCode(200).build()) + .responseBody(AbortableInputStream.create(responseBody)) + .build()); + when(responseHandler.handle(any(), any())).thenReturn(VoidSdkResponse.builder().build()); + } + + @Test + public void responseInterceptorFailureClosesResponseBody() throws Exception { + assertThatThrownBy(() -> clientHandler.execute(clientExecutionParams())) + .hasMessage(hook.name()); + + verify(responseBody).close(); + } + + private SdkClientConfiguration clientConfiguration() { + return HttpTestUtils.testClientConfiguration().toBuilder() + .option(SdkClientOption.EXECUTION_INTERCEPTORS, Collections.singletonList(hook.interceptor())) + .option(SdkClientOption.SYNC_HTTP_CLIENT, httpClient) + .option(SdkClientOption.RETRY_STRATEGY, DefaultRetryStrategy.doNotRetry()) + .build(); + } + + private ClientExecutionParams clientExecutionParams() { + return new ClientExecutionParams() + .withInput(request) + .withMarshaller(marshaller) + .withResponseHandler(responseHandler) + .withErrorResponseHandler(errorResponseHandler); + } + + private enum Hook { + AFTER_TRANSMISSION(new ExecutionInterceptor() { + @Override + public void afterTransmission(Context.AfterTransmission context, ExecutionAttributes executionAttributes) { + throw new RuntimeException(AFTER_TRANSMISSION.name()); + } + }), + + MODIFY_HTTP_RESPONSE(new ExecutionInterceptor() { + @Override + public SdkHttpResponse modifyHttpResponse(Context.ModifyHttpResponse context, + ExecutionAttributes executionAttributes) { + throw new RuntimeException(MODIFY_HTTP_RESPONSE.name()); + } + }), + + MODIFY_HTTP_RESPONSE_CONTENT(new ExecutionInterceptor() { + @Override + public Optional modifyHttpResponseContent(Context.ModifyHttpResponse context, + ExecutionAttributes executionAttributes) { + throw new RuntimeException(MODIFY_HTTP_RESPONSE_CONTENT.name()); + } + }), + + BEFORE_UNMARSHALLING(new ExecutionInterceptor() { + @Override + public void beforeUnmarshalling(Context.BeforeUnmarshalling context, + ExecutionAttributes executionAttributes) { + throw new RuntimeException(BEFORE_UNMARSHALLING.name()); + } + }); + + private final ExecutionInterceptor interceptor; + + Hook(ExecutionInterceptor interceptor) { + this.interceptor = interceptor; + } + + private ExecutionInterceptor interceptor() { + return interceptor; + } + } +}