Skip to content

Commit e8b30e1

Browse files
afarberafarber
committed
Close HTTP response streams when interceptors throw
1 parent e5dfd85 commit e8b30e1

5 files changed

Lines changed: 216 additions & 16 deletions

File tree

‎.changes/next-release/bugfix-AWSSDKforJavav2-4927ecb.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"type": "bugfix",
3+
"category": "AWS SDK for Java v2",
4+
"contributor": "afarber",
5+
"description": "Close synchronous HTTP response streams when response interceptors throw exceptions."
6+
}

‎core/sdk-core/src/main/java/software/amazon/awssdk/core/interceptor/ExecutionInterceptorChain.java‎

Lines changed: 25 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import software.amazon.awssdk.core.sync.RequestBody;
3131
import software.amazon.awssdk.http.SdkHttpRequest;
3232
import software.amazon.awssdk.http.SdkHttpResponse;
33+
import software.amazon.awssdk.utils.IoUtils;
3334
import software.amazon.awssdk.utils.Logger;
3435
import software.amazon.awssdk.utils.Validate;
3536

@@ -112,22 +113,33 @@ public void afterTransmission(Context.AfterTransmission context, ExecutionAttrib
112113
public InterceptorContext modifyHttpResponse(InterceptorContext context,
113114
ExecutionAttributes executionAttributes) {
114115
InterceptorContext result = context;
115-
116-
for (int i = interceptors.size() - 1; i >= 0; i--) {
117-
SdkHttpResponse interceptorResult =
118-
interceptors.get(i).modifyHttpResponse(result, executionAttributes);
119-
InputStream response = interceptors.get(i).modifyHttpResponseContent(result, executionAttributes).orElse(null);
120-
121-
if (interceptorResult != result.httpResponse() || response != result.responseBody().orElse(null)) {
122-
validateInterceptorResult(result.httpResponse(), interceptorResult, interceptors.get(i), "modifyHttpResponse");
123-
result = result.copy(r -> r.httpResponse(interceptorResult)
124-
.responseBody(response));
116+
InputStream response = context.responseBody().orElse(null);
117+
boolean completed = false;
118+
119+
try {
120+
for (int i = interceptors.size() - 1; i >= 0; i--) {
121+
SdkHttpResponse interceptorResult =
122+
interceptors.get(i).modifyHttpResponse(result, executionAttributes);
123+
response = interceptors.get(i)
124+
.modifyHttpResponseContent(result, executionAttributes)
125+
.orElse(null);
126+
127+
if (interceptorResult != result.httpResponse() || response != result.responseBody().orElse(null)) {
128+
validateInterceptorResult(result.httpResponse(), interceptorResult, interceptors.get(i),
129+
"modifyHttpResponse");
130+
InputStream currentResponse = response;
131+
result = result.copy(r -> r.httpResponse(interceptorResult)
132+
.responseBody(currentResponse));
133+
}
125134
}
126135

127-
136+
completed = true;
137+
return result;
138+
} finally {
139+
if (!completed) {
140+
IoUtils.closeQuietlyV2(response, LOG);
141+
}
128142
}
129-
130-
return result;
131143
}
132144

133145
public InterceptorContext modifyAsyncHttpResponse(InterceptorContext context,

‎core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AfterTransmissionExecutionInterceptorsStage.java‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,15 @@
2323
import software.amazon.awssdk.http.AbortableInputStream;
2424
import software.amazon.awssdk.http.SdkHttpFullRequest;
2525
import software.amazon.awssdk.http.SdkHttpFullResponse;
26+
import software.amazon.awssdk.utils.IoUtils;
27+
import software.amazon.awssdk.utils.Logger;
2628
import software.amazon.awssdk.utils.Pair;
2729

2830
@SdkInternalApi
2931
public class AfterTransmissionExecutionInterceptorsStage
3032
implements RequestPipeline<Pair<SdkHttpFullRequest, SdkHttpFullResponse>, Pair<SdkHttpFullRequest, SdkHttpFullResponse>> {
33+
private static final Logger LOG = Logger.loggerFor(AfterTransmissionExecutionInterceptorsStage.class);
34+
3135
@Override
3236
public Pair<SdkHttpFullRequest, SdkHttpFullResponse> execute(Pair<SdkHttpFullRequest, SdkHttpFullResponse> input,
3337
RequestExecutionContext context) throws Exception {
@@ -40,7 +44,15 @@ public Pair<SdkHttpFullRequest, SdkHttpFullResponse> execute(Pair<SdkHttpFullReq
4044
.orElse(null)));
4145

4246
// interceptors.afterTransmission
43-
context.interceptorChain().afterTransmission(interceptorContext, context.executionAttributes());
47+
boolean completed = false;
48+
try {
49+
context.interceptorChain().afterTransmission(interceptorContext, context.executionAttributes());
50+
completed = true;
51+
} finally {
52+
if (!completed) {
53+
input.right().content().ifPresent(stream -> IoUtils.closeQuietlyV2(stream, LOG));
54+
}
55+
}
4456

4557
// interceptors.modifyHttpResponse
4658
interceptorContext = context.interceptorChain().modifyHttpResponse(interceptorContext, context.executionAttributes());

‎core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/BeforeUnmarshallingExecutionInterceptorsStage.java‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
import software.amazon.awssdk.http.SdkHttpFullRequest;
2424
import software.amazon.awssdk.http.SdkHttpFullResponse;
2525
import software.amazon.awssdk.http.SdkHttpResponse;
26+
import software.amazon.awssdk.utils.IoUtils;
27+
import software.amazon.awssdk.utils.Logger;
2628
import software.amazon.awssdk.utils.Pair;
2729

2830
/**
@@ -32,12 +34,21 @@
3234
@SdkInternalApi
3335
public class BeforeUnmarshallingExecutionInterceptorsStage
3436
implements RequestPipeline<Pair<SdkHttpFullRequest, SdkHttpFullResponse>, SdkHttpFullResponse> {
37+
private static final Logger LOG = Logger.loggerFor(BeforeUnmarshallingExecutionInterceptorsStage.class);
3538

3639
@Override
3740
public SdkHttpFullResponse execute(Pair<SdkHttpFullRequest, SdkHttpFullResponse> input,
3841
RequestExecutionContext context) throws Exception {
39-
context.interceptorChain().beforeUnmarshalling(context.executionContext().interceptorContext(),
40-
context.executionAttributes());
42+
boolean completed = false;
43+
try {
44+
context.interceptorChain().beforeUnmarshalling(context.executionContext().interceptorContext(),
45+
context.executionAttributes());
46+
completed = true;
47+
} finally {
48+
if (!completed) {
49+
input.right().content().ifPresent(stream -> IoUtils.closeQuietlyV2(stream, LOG));
50+
}
51+
}
4152
InterruptMonitor.checkInterrupted(input.right());
4253
return input.right();
4354
}

‎core/sdk-core/src/test/java/software/amazon/awssdk/core/client/handler/SyncClientHandlerInterceptorExceptionTest.java‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
/*
2+
* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License").
5+
* You may not use this file except in compliance with the License.
6+
* A copy of the License is located at
7+
*
8+
* http://aws.amazon.com/apache2.0
9+
*
10+
* or in the "license" file. This file is distributed on an "AS IS" BASIS,
11+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
* See the License for the specific language governing permissions and
13+
* limitations under the License.
14+
*/
15+
16+
package software.amazon.awssdk.core.client.handler;
17+
18+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
19+
import static org.mockito.ArgumentMatchers.any;
20+
import static org.mockito.Mockito.mock;
21+
import static org.mockito.Mockito.verify;
22+
import static org.mockito.Mockito.when;
23+
24+
import java.io.InputStream;
25+
import java.util.Arrays;
26+
import java.util.Collection;
27+
import java.util.Collections;
28+
import java.util.Optional;
29+
import java.util.stream.Collectors;
30+
import org.junit.Before;
31+
import org.junit.Test;
32+
import org.junit.runner.RunWith;
33+
import org.junit.runners.Parameterized;
34+
import software.amazon.awssdk.core.SdkRequest;
35+
import software.amazon.awssdk.core.SdkResponse;
36+
import software.amazon.awssdk.core.client.config.SdkClientConfiguration;
37+
import software.amazon.awssdk.core.client.config.SdkClientOption;
38+
import software.amazon.awssdk.core.exception.SdkServiceException;
39+
import software.amazon.awssdk.core.http.HttpResponseHandler;
40+
import software.amazon.awssdk.core.interceptor.Context;
41+
import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
42+
import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
43+
import software.amazon.awssdk.core.protocol.VoidSdkResponse;
44+
import software.amazon.awssdk.core.runtime.transform.Marshaller;
45+
import software.amazon.awssdk.http.AbortableInputStream;
46+
import software.amazon.awssdk.http.ExecutableHttpRequest;
47+
import software.amazon.awssdk.http.HttpExecuteResponse;
48+
import software.amazon.awssdk.http.SdkHttpClient;
49+
import software.amazon.awssdk.http.SdkHttpFullRequest;
50+
import software.amazon.awssdk.http.SdkHttpResponse;
51+
import software.amazon.awssdk.retries.DefaultRetryStrategy;
52+
import utils.HttpTestUtils;
53+
import utils.ValidSdkObjects;
54+
55+
@RunWith(Parameterized.class)
56+
public class SyncClientHandlerInterceptorExceptionTest {
57+
private final SdkRequest request = mock(SdkRequest.class);
58+
private final SdkHttpClient httpClient = mock(SdkHttpClient.class);
59+
private final ExecutableHttpRequest httpClientCall = mock(ExecutableHttpRequest.class);
60+
private final Marshaller<SdkRequest> marshaller = mock(Marshaller.class);
61+
private final HttpResponseHandler<SdkResponse> responseHandler = mock(HttpResponseHandler.class);
62+
private final HttpResponseHandler<SdkServiceException> errorResponseHandler = mock(HttpResponseHandler.class);
63+
private final InputStream responseBody = mock(InputStream.class);
64+
65+
private final Hook hook;
66+
private SdkSyncClientHandler clientHandler;
67+
68+
@Parameterized.Parameters(name = "Interceptor Hook: {0}")
69+
public static Collection<Object[]> data() {
70+
return Arrays.stream(Hook.values())
71+
.map(hook -> new Object[] {hook})
72+
.collect(Collectors.toList());
73+
}
74+
75+
public SyncClientHandlerInterceptorExceptionTest(Hook hook) {
76+
this.hook = hook;
77+
}
78+
79+
@Before
80+
public void setUp() throws Exception {
81+
clientHandler = new SdkSyncClientHandler(clientConfiguration());
82+
83+
when(request.overrideConfiguration()).thenReturn(Optional.empty());
84+
when(marshaller.marshall(request)).thenReturn(ValidSdkObjects.sdkHttpFullRequest().build());
85+
when(httpClient.prepareRequest(any())).thenReturn(httpClientCall);
86+
when(httpClientCall.call()).thenReturn(HttpExecuteResponse.builder()
87+
.response(SdkHttpResponse.builder().statusCode(200).build())
88+
.responseBody(AbortableInputStream.create(responseBody))
89+
.build());
90+
when(responseHandler.handle(any(), any())).thenReturn(VoidSdkResponse.builder().build());
91+
}
92+
93+
@Test
94+
public void responseInterceptorFailureClosesResponseBody() throws Exception {
95+
assertThatThrownBy(() -> clientHandler.execute(clientExecutionParams()))
96+
.hasMessage(hook.name());
97+
98+
verify(responseBody).close();
99+
}
100+
101+
private SdkClientConfiguration clientConfiguration() {
102+
return HttpTestUtils.testClientConfiguration().toBuilder()
103+
.option(SdkClientOption.EXECUTION_INTERCEPTORS, Collections.singletonList(hook.interceptor()))
104+
.option(SdkClientOption.SYNC_HTTP_CLIENT, httpClient)
105+
.option(SdkClientOption.RETRY_STRATEGY, DefaultRetryStrategy.doNotRetry())
106+
.build();
107+
}
108+
109+
private ClientExecutionParams<SdkRequest, SdkResponse> clientExecutionParams() {
110+
return new ClientExecutionParams<SdkRequest, SdkResponse>()
111+
.withInput(request)
112+
.withMarshaller(marshaller)
113+
.withResponseHandler(responseHandler)
114+
.withErrorResponseHandler(errorResponseHandler);
115+
}
116+
117+
private enum Hook {
118+
AFTER_TRANSMISSION(new ExecutionInterceptor() {
119+
@Override
120+
public void afterTransmission(Context.AfterTransmission context, ExecutionAttributes executionAttributes) {
121+
throw new RuntimeException(AFTER_TRANSMISSION.name());
122+
}
123+
}),
124+
125+
MODIFY_HTTP_RESPONSE(new ExecutionInterceptor() {
126+
@Override
127+
public SdkHttpResponse modifyHttpResponse(Context.ModifyHttpResponse context,
128+
ExecutionAttributes executionAttributes) {
129+
throw new RuntimeException(MODIFY_HTTP_RESPONSE.name());
130+
}
131+
}),
132+
133+
MODIFY_HTTP_RESPONSE_CONTENT(new ExecutionInterceptor() {
134+
@Override
135+
public Optional<InputStream> modifyHttpResponseContent(Context.ModifyHttpResponse context,
136+
ExecutionAttributes executionAttributes) {
137+
throw new RuntimeException(MODIFY_HTTP_RESPONSE_CONTENT.name());
138+
}
139+
}),
140+
141+
BEFORE_UNMARSHALLING(new ExecutionInterceptor() {
142+
@Override
143+
public void beforeUnmarshalling(Context.BeforeUnmarshalling context,
144+
ExecutionAttributes executionAttributes) {
145+
throw new RuntimeException(BEFORE_UNMARSHALLING.name());
146+
}
147+
});
148+
149+
private final ExecutionInterceptor interceptor;
150+
151+
Hook(ExecutionInterceptor interceptor) {
152+
this.interceptor = interceptor;
153+
}
154+
155+
private ExecutionInterceptor interceptor() {
156+
return interceptor;
157+
}
158+
}
159+
}

0 commit comments

Comments
 (0)