Skip to content

Commit c984cff

Browse files
manusamanusa
committed
refactor: reduce StandardHttpClient retry complexity
- Abstracts the retryWithExponentialBackoff method so that it can be reused - Provides specific testing for all scenarios Signed-off-by: Marc Nuri <[email protected]>
1 parent 6d7db80 commit c984cff

File tree

4 files changed

+302
-72
lines changed

4 files changed

+302
-72
lines changed

‎kubernetes-client-api/src/main/java/io/fabric8/kubernetes/client/http/StandardHttpClient.java‎

Lines changed: 47 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import io.fabric8.kubernetes.client.http.AsyncBody.Consumer;
2121
import io.fabric8.kubernetes.client.http.Interceptor.RequestTags;
2222
import io.fabric8.kubernetes.client.http.WebSocket.Listener;
23+
import io.fabric8.kubernetes.client.utils.AsyncUtils;
2324
import io.fabric8.kubernetes.client.utils.ExponentialBackoffIntervalCalculator;
2425
import io.fabric8.kubernetes.client.utils.Utils;
2526
import org.slf4j.Logger;
@@ -34,18 +35,16 @@
3435
import java.util.Optional;
3536
import java.util.concurrent.CompletableFuture;
3637
import java.util.concurrent.CompletionException;
37-
import java.util.concurrent.Future;
3838
import java.util.concurrent.TimeUnit;
39-
import java.util.concurrent.TimeoutException;
4039
import java.util.function.BiConsumer;
41-
import java.util.function.Function;
4240
import java.util.function.Supplier;
41+
import java.util.function.ToIntFunction;
4342

4443
public abstract class StandardHttpClient<C extends HttpClient, F extends HttpClient.Factory, T extends StandardHttpClientBuilder<C, F, ?>>
4544
implements HttpClient, RequestTags {
4645

4746
// pads the fail-safe timeout to ensure we don't inadvertently timeout a request
48-
private static final long ADDITIONAL_REQEUST_TIMEOUT = TimeUnit.SECONDS.toMillis(5);
47+
private static final long MAX_ADDITIONAL_REQUEST_TIMEOUT = TimeUnit.SECONDS.toMillis(5);
4948

5049
private static final Logger LOG = LoggerFactory.getLogger(StandardHttpClient.class);
5150

@@ -81,13 +80,12 @@ public <V> CompletableFuture<HttpResponse<V>> sendAsync(HttpRequest request, Cla
8180

8281
@Override
8382
public CompletableFuture<HttpResponse<AsyncBody>> consumeBytes(HttpRequest request, Consumer<List<ByteBuffer>> consumer) {
84-
CompletableFuture<HttpResponse<AsyncBody>> result = new CompletableFuture<>();
85-
StandardHttpRequest standardHttpRequest = (StandardHttpRequest) request;
86-
87-
retryWithExponentialBackoff(result, () -> consumeBytesOnce(standardHttpRequest, consumer), request.uri(),
88-
HttpResponse::code,
89-
r -> r.body().cancel(), standardHttpRequest.getTimeout());
90-
return result;
83+
final StandardHttpRequest standardHttpRequest = (StandardHttpRequest) request;
84+
return retryWithExponentialBackoff(
85+
standardHttpRequest,
86+
() -> consumeBytesOnce(standardHttpRequest, consumer),
87+
r -> r.body().cancel(),
88+
HttpResponse::code);
9189
}
9290

9391
private CompletableFuture<HttpResponse<AsyncBody>> consumeBytesOnce(StandardHttpRequest standardHttpRequest,
@@ -143,69 +141,48 @@ private CompletableFuture<HttpResponse<AsyncBody>> consumeBytesOnce(StandardHttp
143141
};
144142
}
145143

146-
public <V> CompletableFuture<V> orTimeout(CompletableFuture<V> future, Duration timeout) {
147-
if (timeout != null && !timeout.isNegative() && !timeout.isZero()) {
148-
long millis = timeout.toMillis();
149-
millis += (Math.min(millis, ADDITIONAL_REQEUST_TIMEOUT));
150-
Future<?> scheduled = Utils.schedule(Runnable::run, () -> future.completeExceptionally(new TimeoutException()),
151-
millis, TimeUnit.MILLISECONDS);
152-
future.whenComplete((v, t) -> scheduled.cancel(true));
153-
}
154-
return future;
155-
}
156-
157144
/**
158145
* Will retry the action if needed based upon the retry settings provided by the ExponentialBackoffIntervalCalculator.
159146
*/
160-
protected <V> void retryWithExponentialBackoff(CompletableFuture<V> result,
161-
Supplier<CompletableFuture<V>> action, URI uri, Function<V, Integer> codeExtractor,
162-
java.util.function.Consumer<V> cancel, ExponentialBackoffIntervalCalculator retryIntervalCalculator,
163-
Duration timeout) {
164-
165-
orTimeout(action.get(), timeout)
166-
.whenComplete((response, throwable) -> {
167-
if (retryIntervalCalculator.shouldRetry() && !result.isDone()) {
168-
long retryInterval = retryIntervalCalculator.nextReconnectInterval();
169-
boolean retry = false;
170-
if (response != null) {
171-
Integer code = codeExtractor.apply(response);
172-
if (code != null && code >= 500) {
173-
LOG.debug("HTTP operation on url: {} should be retried as the response code was {}, retrying after {} millis",
174-
uri, code, retryInterval);
175-
retry = true;
176-
cancel.accept(response);
177-
}
178-
} else {
179-
if (throwable instanceof CompletionException) {
180-
throwable = ((CompletionException) throwable).getCause();
181-
}
182-
if (throwable instanceof IOException) {
183-
LOG.debug(String.format("HTTP operation on url: %s should be retried after %d millis because of IOException",
184-
uri, retryInterval), throwable);
185-
retry = true;
186-
}
147+
private <V> CompletableFuture<V> retryWithExponentialBackoff(
148+
StandardHttpRequest request, Supplier<CompletableFuture<V>> action, java.util.function.Consumer<V> onCancel,
149+
ToIntFunction<V> codeExtractor) {
150+
final URI uri = request.uri();
151+
final RequestConfig requestConfig = getTag(RequestConfig.class);
152+
final ExponentialBackoffIntervalCalculator retryIntervalCalculator = ExponentialBackoffIntervalCalculator
153+
.from(requestConfig);
154+
final Duration timeout;
155+
if (request.getTimeout() != null && !request.getTimeout().isNegative() && !request.getTimeout().isZero()) {
156+
timeout = request.getTimeout().plusMillis(Math.min(request.getTimeout().toMillis(), MAX_ADDITIONAL_REQUEST_TIMEOUT));
157+
} else {
158+
timeout = null;
159+
}
160+
return AsyncUtils.retryWithExponentialBackoff(action, onCancel, timeout, retryIntervalCalculator,
161+
(response, throwable, retryInterval) -> {
162+
if (response != null) {
163+
final int code = codeExtractor.applyAsInt(response);
164+
if (code >= 500) {
165+
LOG.debug(
166+
"HTTP operation on url: {} should be retried as the response code was {}, retrying after {} millis",
167+
uri, code, retryInterval);
168+
return true;
169+
}
170+
} else {
171+
if (throwable instanceof CompletionException) {
172+
throwable = throwable.getCause();
187173
}
188-
if (retry) {
189-
Utils.schedule(Runnable::run,
190-
() -> retryWithExponentialBackoff(result, action, uri, codeExtractor, cancel, retryIntervalCalculator,
191-
timeout),
192-
retryInterval,
193-
TimeUnit.MILLISECONDS);
194-
return;
174+
if (throwable instanceof IOException) {
175+
LOG.debug(
176+
String.format("HTTP operation on url: %s should be retried after %d millis because of IOException",
177+
uri, retryInterval),
178+
throwable);
179+
return true;
195180
}
196181
}
197-
completeOrCancel(cancel, result).accept(response, throwable);
182+
return false;
198183
});
199184
}
200185

201-
protected <V> void retryWithExponentialBackoff(CompletableFuture<V> result,
202-
Supplier<CompletableFuture<V>> action, URI uri, Function<V, Integer> codeExtractor,
203-
java.util.function.Consumer<V> cancel, Duration timeout) {
204-
RequestConfig requestConfig = getTag(RequestConfig.class);
205-
retryWithExponentialBackoff(result, action, uri, codeExtractor, cancel,
206-
ExponentialBackoffIntervalCalculator.from(requestConfig), timeout);
207-
}
208-
209186
@Override
210187
public io.fabric8.kubernetes.client.http.WebSocket.Builder newWebSocketBuilder() {
211188
return new StandardWebSocketBuilder(this);
@@ -220,13 +197,11 @@ public HttpRequest.Builder newHttpRequestBuilder() {
220197
final CompletableFuture<WebSocket> buildWebSocket(StandardWebSocketBuilder standardWebSocketBuilder,
221198
Listener listener) {
222199

223-
CompletableFuture<WebSocketResponse> intermediate = new CompletableFuture<>();
224-
StandardHttpRequest request = standardWebSocketBuilder.asHttpRequest();
225-
226-
retryWithExponentialBackoff(intermediate, () -> buildWebSocketOnce(standardWebSocketBuilder, listener),
227-
request.uri(),
228-
r -> Optional.of(r.webSocketUpgradeResponse).map(HttpResponse::code).orElse(null),
229-
r -> Optional.ofNullable(r.webSocket).ifPresent(w -> w.sendClose(1000, null)), request.getTimeout());
200+
final CompletableFuture<WebSocketResponse> intermediate = retryWithExponentialBackoff(
201+
standardWebSocketBuilder.asHttpRequest(),
202+
() -> buildWebSocketOnce(standardWebSocketBuilder, listener),
203+
r -> Optional.ofNullable(r.webSocket).ifPresent(w -> w.sendClose(1000, null)),
204+
r -> Optional.of(r.webSocketUpgradeResponse).map(HttpResponse::code).orElse(null));
230205

231206
CompletableFuture<WebSocket> result = new CompletableFuture<>();
232207

‎kubernetes-client-api/src/main/java/io/fabric8/kubernetes/client/utils/AsyncUtils.java‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/**
2+
* Copyright (C) 2015 Red Hat, Inc.
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+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.fabric8.kubernetes.client.utils;
17+
18+
import java.time.Duration;
19+
import java.util.concurrent.CompletableFuture;
20+
import java.util.concurrent.Future;
21+
import java.util.concurrent.TimeUnit;
22+
import java.util.concurrent.TimeoutException;
23+
import java.util.function.Consumer;
24+
import java.util.function.Supplier;
25+
26+
public class AsyncUtils {
27+
28+
private AsyncUtils() {
29+
}
30+
31+
/**
32+
* Returns the provided {@link CompletableFuture} that will complete exceptionally with a {@link TimeoutException}
33+
* if the provided {@link Duration} timeout period is exceeded.
34+
*
35+
* @param future the future to add a timeout to.
36+
* @param timeout the timeout duration.
37+
* @return the provided future with a timeout.
38+
* @param <T> the result type returned by the future.
39+
*/
40+
public static <T> CompletableFuture<T> withTimeout(CompletableFuture<T> future, Duration timeout) {
41+
if (timeout != null && timeout.toMillis() > 0) {
42+
final Future<?> scheduled = Utils.schedule(Runnable::run, () -> future.completeExceptionally(new TimeoutException()),
43+
timeout.toMillis(), TimeUnit.MILLISECONDS);
44+
future.whenComplete((v, t) -> scheduled.cancel(true));
45+
}
46+
return future;
47+
}
48+
49+
/**
50+
* Returns a new {@link CompletableFuture} that will complete once the action provided by the action supplier completes.
51+
* The action will be retried with an exponential backoff using the {@link ExponentialBackoffIntervalCalculator} as
52+
* long as the {@link ShouldRetry} predicate returns true.
53+
* Each action retrieval retry will time out after the provided timeout {@link Duration}.
54+
*
55+
* @param action the action supplier.
56+
* @param onCancel consumes the intermediate result in case the returned future is cancelled or each time the action is
57+
* retried.
58+
* @param timeout the timeout duration.
59+
* @param retryIntervalCalculator the retry interval calculator.
60+
* @param shouldRetry the predicate to compute if the action is to be retried.
61+
* @return a new {@link CompletableFuture} that will complete once the action provided by the action supplier completes.
62+
* @param <T> the result type returned by the future.
63+
*/
64+
public static <T> CompletableFuture<T> retryWithExponentialBackoff(Supplier<CompletableFuture<T>> action,
65+
Consumer<T> onCancel, Duration timeout, ExponentialBackoffIntervalCalculator retryIntervalCalculator,
66+
ShouldRetry<T> shouldRetry) {
67+
final CompletableFuture<T> result = new CompletableFuture<>();
68+
retryWithExponentialBackoff(result, action, onCancel, timeout, retryIntervalCalculator, shouldRetry);
69+
return result;
70+
}
71+
72+
private static <T> void retryWithExponentialBackoff(CompletableFuture<T> result, Supplier<CompletableFuture<T>> action,
73+
Consumer<T> onCancel, Duration timeout, ExponentialBackoffIntervalCalculator retryIntervalCalculator,
74+
ShouldRetry<T> shouldRetry) {
75+
withTimeout(action.get(), timeout).whenComplete((r, t) -> {
76+
if (retryIntervalCalculator.shouldRetry() && !result.isDone()) {
77+
final long retryInterval = retryIntervalCalculator.nextReconnectInterval();
78+
if (shouldRetry.shouldRetry(r, t, retryInterval)) {
79+
if (r != null) {
80+
onCancel.accept(r);
81+
}
82+
Utils.schedule(Runnable::run,
83+
() -> retryWithExponentialBackoff(result, action, onCancel, timeout, retryIntervalCalculator, shouldRetry),
84+
retryInterval, TimeUnit.MILLISECONDS);
85+
return;
86+
}
87+
}
88+
if (t != null) {
89+
result.completeExceptionally(t);
90+
} else if (!result.complete(r)) {
91+
onCancel.accept(r);
92+
}
93+
});
94+
}
95+
96+
@FunctionalInterface
97+
public interface ShouldRetry<T> {
98+
boolean shouldRetry(T result, Throwable exception, long retryInterval);
99+
}
100+
}

‎kubernetes-client-api/src/test/java/io/fabric8/kubernetes/client/http/StandardHttpClientTest.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import java.util.concurrent.CompletableFuture;
2929
import java.util.concurrent.ExecutionException;
3030
import java.util.concurrent.TimeUnit;
31+
import java.util.concurrent.TimeoutException;
3132
import java.util.stream.IntStream;
3233

3334
import static org.assertj.core.api.Assertions.assertThat;
@@ -221,6 +222,8 @@ void testRequestTimeout() {
221222
});
222223

223224
Awaitility.await().atMost(10, TimeUnit.SECONDS).until(consumeFuture::isDone);
225+
assertThatThrownBy(consumeFuture::get)
226+
.isInstanceOf(ExecutionException.class).hasCauseInstanceOf(TimeoutException.class);
224227
}
225228

226229
}

0 commit comments

Comments
 (0)