prateekm commented on code in PR #1650:
URL: https://github.com/apache/samza/pull/1650#discussion_r1092604327
##########
samza-core/src/main/java/org/apache/samza/util/FutureUtil.java:
##########
@@ -143,20 +143,32 @@ public static <K, V> CompletableFuture<Map<K, V>>
toFutureOfMap(
}
public static <T> CompletableFuture<T> executeAsyncWithRetries(String opName,
- Supplier<? extends CompletionStage<T>> action,
- Predicate<? extends Throwable> abortRetries,
+ Supplier<? extends CompletionStage<T>> action, Predicate<? extends
Throwable> abortRetries,
ExecutorService executor) {
Duration maxDuration = Duration.ofMinutes(10);
RetryPolicy<Object> retryPolicy = new RetryPolicy<>()
.withMaxRetries(-1) // Sets maximum retry to unlimited from default of
3 attempts. Retries are now limited by max duration and not retry counts.
.withBackoff(100, 312500, ChronoUnit.MILLIS, 5) // 100 ms, 500 ms,
2500 ms, 12.5 s, 1.05 min, 5.20 min, 5.20 min
.withMaxDuration(maxDuration)
+ .withJitter(Duration.ofMillis(100))
.abortOn(abortRetries) // stop retrying if predicate returns true
.onRetry(e -> LOG.warn("Action: {} attempt: {} completed with error {}
ms after start. Retrying up to {} ms.",
opName, e.getAttemptCount(), e.getElapsedTime().toMillis(),
maxDuration.toMillis(), e.getLastFailure()));
- return
Failsafe.with(retryPolicy).with(executor).getStageAsync(action::get);
+ return executeAsyncWithRetries(opName, action, abortRetries, executor,
retryPolicy);
+ }
+
+ public static <T> CompletableFuture<T> executeAsyncWithRetries(String opName,
+ Supplier<? extends CompletionStage<T>> action, Predicate<? extends
Throwable> abortRetries,
+ ExecutorService executor, RetryPolicy<Object> retryPolicy) {
+
+ RetryPolicy<Object> retryPolicyWithLog = retryPolicy.abortOn(abortRetries)
Review Comment:
Looking at the impl, RetryPolicy is a mutable object, this is not safe to
do. Please fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
