Skip to content
Permalink
Browse files
HTTPCLIENT-2152: Fixed handling of unexpected unchecked exception by …
…the async request retry exec interceptor
  • Loading branch information
ok2c committed Apr 27, 2021
1 parent 13137eb commit 09f50cd80c9cabc9f7532b6063fe67695f5484fa
Showing 3 changed files with 337 additions and 16 deletions.
@@ -0,0 +1,138 @@
/*
* ====================================================================
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License 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.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*
*/

package org.apache.hc.client5.testing.async;

import org.apache.hc.core5.function.Resolver;
import org.apache.hc.core5.http.EntityDetails;
import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpException;
import org.apache.hc.core5.http.HttpHeaders;
import org.apache.hc.core5.http.HttpRequest;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.HttpVersion;
import org.apache.hc.core5.http.ProtocolVersion;
import org.apache.hc.core5.http.message.BasicHttpResponse;
import org.apache.hc.core5.http.nio.AsyncServerExchangeHandler;
import org.apache.hc.core5.http.nio.CapacityChannel;
import org.apache.hc.core5.http.nio.DataStreamChannel;
import org.apache.hc.core5.http.nio.ResponseChannel;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.util.Args;
import org.apache.hc.core5.util.TimeValue;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public class ServiceUnavailableAsyncDecorator implements AsyncServerExchangeHandler {

private final AsyncServerExchangeHandler exchangeHandler;
private final Resolver<HttpRequest, TimeValue> serviceAvailabilityResolver;
private final AtomicBoolean serviceUnavailable;

public ServiceUnavailableAsyncDecorator(final AsyncServerExchangeHandler exchangeHandler,
final Resolver<HttpRequest, TimeValue> serviceAvailabilityResolver) {
this.exchangeHandler = Args.notNull(exchangeHandler, "Exchange handler");
this.serviceAvailabilityResolver = Args.notNull(serviceAvailabilityResolver, "Service availability resolver");
this.serviceUnavailable = new AtomicBoolean();
}

@Override
public void handleRequest(final HttpRequest request,
final EntityDetails entityDetails,
final ResponseChannel responseChannel,
final HttpContext context) throws HttpException, IOException {
final TimeValue retryAfter = serviceAvailabilityResolver.resolve(request);
serviceUnavailable.set(TimeValue.isPositive(retryAfter));
if (serviceUnavailable.get()) {
final HttpResponse response = new BasicHttpResponse(HttpStatus.SC_SERVICE_UNAVAILABLE);
response.addHeader(HttpHeaders.RETRY_AFTER, Long.toString(retryAfter.toSeconds()));
final ProtocolVersion version = request.getVersion();
if (version != null && version.compareToVersion(HttpVersion.HTTP_2) < 0) {
response.addHeader(HttpHeaders.CONNECTION, "Close");
}
responseChannel.sendResponse(response, null, context);
} else {
exchangeHandler.handleRequest(request, entityDetails, responseChannel, context);
}
}

@Override
public final void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
if (!serviceUnavailable.get()) {
exchangeHandler.updateCapacity(capacityChannel);
} else {
capacityChannel.update(Integer.MAX_VALUE);
}
}

@Override
public final void consume(final ByteBuffer src) throws IOException {
if (!serviceUnavailable.get()) {
exchangeHandler.consume(src);
}
}

@Override
public final void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
if (!serviceUnavailable.get()) {
exchangeHandler.streamEnd(trailers);
}
}

@Override
public int available() {
if (!serviceUnavailable.get()) {
return exchangeHandler.available();
} else {
return 0;
}
}

@Override
public void produce(final DataStreamChannel channel) throws IOException {
if (!serviceUnavailable.get()) {
exchangeHandler.produce(channel);
}
}

@Override
public void failed(final Exception cause) {
if (!serviceUnavailable.get()) {
exchangeHandler.failed(cause);
}
}

@Override
public void releaseResources() {
exchangeHandler.releaseResources();
}

}
@@ -0,0 +1,187 @@
/*
* ====================================================================
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License 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.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*
*/
package org.apache.hc.client5.testing.async;

import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
import org.apache.hc.client5.http.config.RequestConfig;
import org.apache.hc.client5.http.impl.DefaultHttpRequestRetryStrategy;
import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
import org.apache.hc.client5.http.impl.async.HttpAsyncClientBuilder;
import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
import org.apache.hc.core5.function.Decorator;
import org.apache.hc.core5.function.Resolver;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.HttpRequest;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.HttpVersion;
import org.apache.hc.core5.http.URIScheme;
import org.apache.hc.core5.http.config.Http1Config;
import org.apache.hc.core5.http.nio.AsyncServerExchangeHandler;
import org.apache.hc.core5.http2.HttpVersionPolicy;
import org.apache.hc.core5.http2.config.H2Config;
import org.apache.hc.core5.util.TimeValue;
import org.hamcrest.CoreMatchers;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExternalResource;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

@RunWith(Parameterized.class)
public class TestHttp1RequestReExecution extends AbstractIntegrationTestBase<CloseableHttpAsyncClient> {

@Parameterized.Parameters(name = "{0}")
public static Collection<Object[]> protocolVersions() {
return Arrays.asList(new Object[][]{
{ HttpVersion.HTTP_1_1 },
{ HttpVersion.HTTP_2 }
});
}

private final HttpVersion version;

public TestHttp1RequestReExecution(final HttpVersion version) {
super(URIScheme.HTTP);
this.version = version;
}

HttpAsyncClientBuilder clientBuilder;
PoolingAsyncClientConnectionManager connManager;

@Rule
public ExternalResource connManagerResource = new ExternalResource() {

@Override
protected void before() throws Throwable {
connManager = PoolingAsyncClientConnectionManagerBuilder.create()
.build();
}

@Override
protected void after() {
if (connManager != null) {
connManager.close();
connManager = null;
}
}

};

@Rule
public ExternalResource clientBuilderResource = new ExternalResource() {

@Override
protected void before() throws Throwable {
clientBuilder = HttpAsyncClientBuilder.create()
.setDefaultRequestConfig(RequestConfig.custom()
.setConnectionRequestTimeout(TIMEOUT)
.setConnectTimeout(TIMEOUT)
.build())
.setConnectionManager(connManager)
.setVersionPolicy(version.greaterEquals(HttpVersion.HTTP_2) ? HttpVersionPolicy.FORCE_HTTP_2 : HttpVersionPolicy.FORCE_HTTP_1);
}

};

@Override
public final HttpHost start() throws Exception {

final Resolver<HttpRequest, TimeValue> serviceAvailabilityResolver = new Resolver<HttpRequest, TimeValue>() {

private final AtomicInteger count = new AtomicInteger(0);

@Override
public TimeValue resolve(final HttpRequest request) {
final int n = count.incrementAndGet();
return n <= 3 ? TimeValue.ofSeconds(1) : null;
}

};

if (version.greaterEquals(HttpVersion.HTTP_2)) {
return super.start(null, new Decorator<AsyncServerExchangeHandler>() {

@Override
public AsyncServerExchangeHandler decorate(final AsyncServerExchangeHandler handler) {
return new ServiceUnavailableAsyncDecorator(handler, serviceAvailabilityResolver);
}

}, H2Config.DEFAULT);
} else {
return super.start(null, new Decorator<AsyncServerExchangeHandler>() {

@Override
public AsyncServerExchangeHandler decorate(final AsyncServerExchangeHandler handler) {
return new ServiceUnavailableAsyncDecorator(handler, serviceAvailabilityResolver);
}

}, Http1Config.DEFAULT);
}
}

@Override
protected CloseableHttpAsyncClient createClient() throws Exception {
return clientBuilder.build();
}

@Test
public void testGiveUpAfterOneRetry() throws Exception {
clientBuilder.setRetryStrategy(new DefaultHttpRequestRetryStrategy(1, TimeValue.ofSeconds(1)));
final HttpHost target = start();
final Future<SimpleHttpResponse> future = httpclient.execute(
SimpleRequestBuilder.get()
.setHttpHost(target)
.setPath("/random/2048")
.build(), null);
final SimpleHttpResponse response = future.get();
Assert.assertThat(response, CoreMatchers.notNullValue());
Assert.assertThat(response.getCode(), CoreMatchers.equalTo(HttpStatus.SC_SERVICE_UNAVAILABLE));
}

@Test
public void testDoNotGiveUpEasily() throws Exception {
clientBuilder.setRetryStrategy(new DefaultHttpRequestRetryStrategy(5, TimeValue.ofSeconds(1)));
final HttpHost target = start();
final Future<SimpleHttpResponse> future = httpclient.execute(
SimpleRequestBuilder.get()
.setHttpHost(target)
.setPath("/random/2048")
.build(), null);
final SimpleHttpResponse response = future.get();
Assert.assertThat(response, CoreMatchers.notNullValue());
Assert.assertThat(response.getCode(), CoreMatchers.equalTo(HttpStatus.SC_OK));
}

}

0 comments on commit 09f50cd

Please sign in to comment.