HTTPCLIENT-2291: fixed inconsistency in behavior between the class and async implementation of the request re-execution. The async request retry exec will now re-start request execution from itself instead of form the very beginning of the execution chain

This commit is contained in:
Oleg Kalnichevski 2023-08-30 19:41:11 +02:00
parent 9748d1baf8
commit 8a54e70359
3 changed files with 75 additions and 8 deletions

View File

@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hc.client5.http.HttpRoute; import org.apache.hc.client5.http.HttpRoute;
import org.apache.hc.client5.http.protocol.HttpClientContext; import org.apache.hc.client5.http.protocol.HttpClientContext;
import org.apache.hc.core5.annotation.Contract; import org.apache.hc.core5.annotation.Contract;
import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.annotation.ThreadingBehavior; import org.apache.hc.core5.annotation.ThreadingBehavior;
import org.apache.hc.core5.concurrent.CancellableDependency; import org.apache.hc.core5.concurrent.CancellableDependency;
import org.apache.hc.core5.http.HttpException; import org.apache.hc.core5.http.HttpException;
@ -116,7 +117,7 @@ public interface AsyncExecChain {
* @param request the actual request. * @param request the actual request.
* @param entityProducer the request entity producer or {@code null} if the request * @param entityProducer the request entity producer or {@code null} if the request
* does not enclose an entity. * does not enclose an entity.
* @param scope the execution scope . * @param scope the execution scope.
* @param asyncExecCallback the execution callback. * @param asyncExecCallback the execution callback.
* @param delay re-execution delay. Can be {@code null} if the request is to be * @param delay re-execution delay. Can be {@code null} if the request is to be
* re-executed immediately. * re-executed immediately.
@ -128,6 +129,31 @@ public interface AsyncExecChain {
AsyncExecCallback asyncExecCallback, AsyncExecCallback asyncExecCallback,
TimeValue delay); TimeValue delay);
/**
* Schedules request re-execution of the given execution chain immediately or
* after a delay.
* @param request the actual request.
* @param entityProducer the request entity producer or {@code null} if the request
* does not enclose an entity.
* @param scope the execution scope.
* @param chain the execution chain.
* @param asyncExecCallback the execution callback.
* @param delay re-execution delay. Can be {@code null} if the request is to be
* re-executed immediately.
*
* @since 5.3
*/
@Internal
default void scheduleExecution(
HttpRequest request,
AsyncEntityProducer entityProducer,
AsyncExecChain.Scope scope,
AsyncExecChain chain,
AsyncExecCallback asyncExecCallback,
TimeValue delay) {
scheduleExecution(request, entityProducer, scope, asyncExecCallback, delay);
}
} }
/** /**

View File

@ -131,7 +131,13 @@ public final class AsyncHttpRequestRetryExec implements AsyncExecChainHandler {
if (entityProducer != null) { if (entityProducer != null) {
entityProducer.releaseResources(); entityProducer.releaseResources();
} }
scope.scheduler.scheduleExecution(request, entityProducer, scope, asyncExecCallback, state.delay); scope.scheduler.scheduleExecution(
request,
entityProducer,
scope,
(r, e, s, c) -> execute(r, e, s, chain, c),
asyncExecCallback,
state.delay);
} else { } else {
asyncExecCallback.completed(); asyncExecCallback.completed();
} }
@ -161,7 +167,13 @@ public final class AsyncHttpRequestRetryExec implements AsyncExecChainHandler {
state.retrying = true; state.retrying = true;
final int execCount = scope.execCount.incrementAndGet(); final int execCount = scope.execCount.incrementAndGet();
state.delay = retryStrategy.getRetryInterval(request, (IOException) cause, execCount - 1, clientContext); state.delay = retryStrategy.getRetryInterval(request, (IOException) cause, execCount - 1, clientContext);
scope.scheduler.scheduleExecution(request, entityProducer, scope, asyncExecCallback, state.delay); scope.scheduler.scheduleExecution(
request,
entityProducer,
scope,
(r, e, s, c) -> execute(r, e, s, chain, c),
asyncExecCallback,
state.delay);
return; return;
} }
} }

View File

@ -92,6 +92,7 @@ abstract class InternalAbstractHttpAsyncClient extends AbstractHttpAsyncClientBa
private final RequestConfig defaultConfig; private final RequestConfig defaultConfig;
private final ConcurrentLinkedQueue<Closeable> closeables; private final ConcurrentLinkedQueue<Closeable> closeables;
private final ScheduledExecutorService scheduledExecutorService; private final ScheduledExecutorService scheduledExecutorService;
private final AsyncExecChain.Scheduler scheduler;
InternalAbstractHttpAsyncClient( InternalAbstractHttpAsyncClient(
final DefaultConnectingIOReactor ioReactor, final DefaultConnectingIOReactor ioReactor,
@ -113,6 +114,30 @@ abstract class InternalAbstractHttpAsyncClient extends AbstractHttpAsyncClientBa
this.defaultConfig = defaultConfig; this.defaultConfig = defaultConfig;
this.closeables = closeables != null ? new ConcurrentLinkedQueue<>(closeables) : null; this.closeables = closeables != null ? new ConcurrentLinkedQueue<>(closeables) : null;
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(SCHEDULER_THREAD_FACTORY); this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(SCHEDULER_THREAD_FACTORY);
this.scheduler = new AsyncExecChain.Scheduler() {
@Override
public void scheduleExecution(
final HttpRequest request,
final AsyncEntityProducer entityProducer,
final AsyncExecChain.Scope scope,
final AsyncExecCallback asyncExecCallback,
final TimeValue delay) {
executeScheduled(request, entityProducer, scope, execChain::execute, asyncExecCallback, delay);
}
@Override
public void scheduleExecution(
final HttpRequest request,
final AsyncEntityProducer entityProducer,
final AsyncExecChain.Scope scope,
final AsyncExecChain chain,
final AsyncExecCallback asyncExecCallback,
final TimeValue delay) {
executeScheduled(request, entityProducer, scope, chain, asyncExecCallback, delay);
}
};
} }
@Override @Override
@ -197,8 +222,6 @@ abstract class InternalAbstractHttpAsyncClient extends AbstractHttpAsyncClientBa
} }
final AsyncExecRuntime execRuntime = createAsyncExecRuntime(pushHandlerFactory); final AsyncExecRuntime execRuntime = createAsyncExecRuntime(pushHandlerFactory);
final AsyncExecChain.Scheduler scheduler = this::executeScheduled;
final AsyncExecChain.Scope scope = new AsyncExecChain.Scope(exchangeId, route, request, future, final AsyncExecChain.Scope scope = new AsyncExecChain.Scope(exchangeId, route, request, future,
clientContext, execRuntime, scheduler, new AtomicInteger(1)); clientContext, execRuntime, scheduler, new AtomicInteger(1));
final AtomicBoolean outputTerminated = new AtomicBoolean(false); final AtomicBoolean outputTerminated = new AtomicBoolean(false);
@ -262,6 +285,7 @@ abstract class InternalAbstractHttpAsyncClient extends AbstractHttpAsyncClientBa
} : null, } : null,
scope, scope,
execChain::execute,
new AsyncExecCallback() { new AsyncExecCallback() {
@Override @Override
@ -343,18 +367,20 @@ abstract class InternalAbstractHttpAsyncClient extends AbstractHttpAsyncClientBa
final HttpRequest request, final HttpRequest request,
final AsyncEntityProducer entityProducer, final AsyncEntityProducer entityProducer,
final AsyncExecChain.Scope scope, final AsyncExecChain.Scope scope,
final AsyncExecChain chain,
final AsyncExecCallback asyncExecCallback) throws HttpException, IOException { final AsyncExecCallback asyncExecCallback) throws HttpException, IOException {
execChain.execute(request, entityProducer, scope, asyncExecCallback); chain.proceed(request, entityProducer, scope, asyncExecCallback);
} }
void executeScheduled( void executeScheduled(
final HttpRequest request, final HttpRequest request,
final AsyncEntityProducer entityProducer, final AsyncEntityProducer entityProducer,
final AsyncExecChain.Scope scope, final AsyncExecChain.Scope scope,
final AsyncExecChain chain,
final AsyncExecCallback asyncExecCallback, final AsyncExecCallback asyncExecCallback,
final TimeValue delay) { final TimeValue delay) {
final ScheduledRequestExecution scheduledTask = new ScheduledRequestExecution( final ScheduledRequestExecution scheduledTask = new ScheduledRequestExecution(
request, entityProducer, scope, asyncExecCallback, delay); request, entityProducer, scope, chain, asyncExecCallback, delay);
if (TimeValue.isPositive(delay)) { if (TimeValue.isPositive(delay)) {
scheduledExecutorService.schedule(scheduledTask, delay.getDuration(), delay.getTimeUnit()); scheduledExecutorService.schedule(scheduledTask, delay.getDuration(), delay.getTimeUnit());
} else { } else {
@ -367,17 +393,20 @@ abstract class InternalAbstractHttpAsyncClient extends AbstractHttpAsyncClientBa
final HttpRequest request; final HttpRequest request;
final AsyncEntityProducer entityProducer; final AsyncEntityProducer entityProducer;
final AsyncExecChain.Scope scope; final AsyncExecChain.Scope scope;
final AsyncExecChain chain;
final AsyncExecCallback asyncExecCallback; final AsyncExecCallback asyncExecCallback;
final TimeValue delay; final TimeValue delay;
ScheduledRequestExecution(final HttpRequest request, ScheduledRequestExecution(final HttpRequest request,
final AsyncEntityProducer entityProducer, final AsyncEntityProducer entityProducer,
final AsyncExecChain.Scope scope, final AsyncExecChain.Scope scope,
final AsyncExecChain chain,
final AsyncExecCallback asyncExecCallback, final AsyncExecCallback asyncExecCallback,
final TimeValue delay) { final TimeValue delay) {
this.request = request; this.request = request;
this.entityProducer = entityProducer; this.entityProducer = entityProducer;
this.scope = scope; this.scope = scope;
this.chain = chain;
this.asyncExecCallback = asyncExecCallback; this.asyncExecCallback = asyncExecCallback;
this.delay = delay; this.delay = delay;
} }
@ -385,7 +414,7 @@ abstract class InternalAbstractHttpAsyncClient extends AbstractHttpAsyncClientBa
@Override @Override
public void run() { public void run() {
try { try {
execChain.execute(request, entityProducer, scope, asyncExecCallback); chain.proceed(request, entityProducer, scope, asyncExecCallback);
} catch (final Exception ex) { } catch (final Exception ex) {
asyncExecCallback.failed(ex); asyncExecCallback.failed(ex);
} }