diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java index 6bbd3b6601..d22a48fc9e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java @@ -206,9 +206,11 @@ private class PollLoopTask implements Runnable { this.task = task; this.pollBackoffThrottler = new BackoffThrottler( - pollerOptions.getPollBackoffInitialInterval(), - pollerOptions.getPollBackoffMaximumInterval(), - pollerOptions.getPollBackoffCoefficient()); + pollerOptions.getBackoffInitialInterval(), + pollerOptions.getBackoffCongestionInitialInterval(), + pollerOptions.getBackoffMaximumInterval(), + pollerOptions.getBackoffCoefficient(), + pollerOptions.getBackoffMaximumJitterCoefficient()); } @Override @@ -242,7 +244,10 @@ public void run() { Thread.currentThread().interrupt(); } else { // Don't increase throttle on InterruptedException - pollBackoffThrottler.failure(); + pollBackoffThrottler.failure( + (e instanceof StatusRuntimeException) + ? ((StatusRuntimeException) e).getStatus().getCode() + : Status.Code.UNKNOWN); } uncaughtExceptionHandler.uncaughtException(Thread.currentThread(), e); } finally { diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java index 8cb78af9db..192e29c0f5 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java @@ -54,9 +54,11 @@ public static final class Builder { private int maximumPollRateIntervalMilliseconds = 1000; private double maximumPollRatePerSecond; - private double pollBackoffCoefficient = 2; - private Duration pollBackoffInitialInterval = Duration.ofMillis(100); - private Duration pollBackoffMaximumInterval = Duration.ofMinutes(1); + private double backoffCoefficient = 2; + private Duration backoffInitialInterval = Duration.ofMillis(100); + private Duration backoffCongestionInitialInterval = Duration.ofMillis(1000); + private Duration backoffMaximumInterval = Duration.ofMinutes(1); + private double backoffMaximumJitterCoefficient = 0.1; private int pollThreadCount = 1; private String pollThreadNamePrefix; private Thread.UncaughtExceptionHandler uncaughtExceptionHandler; @@ -69,9 +71,11 @@ private Builder(PollerOptions options) { } this.maximumPollRateIntervalMilliseconds = options.getMaximumPollRateIntervalMilliseconds(); this.maximumPollRatePerSecond = options.getMaximumPollRatePerSecond(); - this.pollBackoffCoefficient = options.getPollBackoffCoefficient(); - this.pollBackoffInitialInterval = options.getPollBackoffInitialInterval(); - this.pollBackoffMaximumInterval = options.getPollBackoffMaximumInterval(); + this.backoffCoefficient = options.getBackoffCoefficient(); + this.backoffInitialInterval = options.getBackoffInitialInterval(); + this.backoffCongestionInitialInterval = options.getBackoffCongestionInitialInterval(); + this.backoffMaximumInterval = options.getBackoffMaximumInterval(); + this.backoffMaximumJitterCoefficient = options.getBackoffMaximumJitterCoefficient(); this.pollThreadCount = options.getPollThreadCount(); this.pollThreadNamePrefix = options.getPollThreadNamePrefix(); this.uncaughtExceptionHandler = options.getUncaughtExceptionHandler(); @@ -93,23 +97,41 @@ public Builder setMaximumPollRatePerSecond(double maximumPollRatePerSecond) { } /** Coefficient to use when calculating exponential delay in case of failures */ - public Builder setPollBackoffCoefficient(double pollBackoffCoefficient) { - this.pollBackoffCoefficient = pollBackoffCoefficient; + public Builder setBackoffCoefficient(double backoffCoefficient) { + this.backoffCoefficient = backoffCoefficient; return this; } /** - * Initial delay in case of failure. If backoff coefficient is 1 then it would be the constant - * delay between failing polls. + * Initial delay in case of regular failure. If backoff coefficient is 1 then it would be the + * constant delay between failing polls. */ - public Builder setPollBackoffInitialInterval(Duration pollBackoffInitialInterval) { - this.pollBackoffInitialInterval = pollBackoffInitialInterval; + public Builder setBackoffInitialInterval(Duration backoffInitialInterval) { + this.backoffInitialInterval = backoffInitialInterval; + return this; + } + + /** + * Initial delay in case of congestion-related failures (i.e. RESOURCE_EXHAUSTED errors). If + * backoff coefficient is 1 then it would be the constant delay between failing polls. + */ + public Builder setBackoffCongestionInitialInterval(Duration backoffCongestionInitialInterval) { + this.backoffCongestionInitialInterval = backoffCongestionInitialInterval; return this; } /** Maximum interval between polls in case of failures. */ - public Builder setPollBackoffMaximumInterval(Duration pollBackoffMaximumInterval) { - this.pollBackoffMaximumInterval = pollBackoffMaximumInterval; + public Builder setBackoffMaximumInterval(Duration backoffMaximumInterval) { + this.backoffMaximumInterval = backoffMaximumInterval; + return this; + } + + /** + * Maximum amount of jitter to apply. 0.2 means that actual retry time can be +/- 20% of the + * calculated time. Set to 0 to disable jitter. Must be lower than 1. Default is 0.1. + */ + public Builder setBackoffMaximumJitterCoefficient(double backoffMaximumJitterCoefficient) { + this.backoffMaximumJitterCoefficient = backoffMaximumJitterCoefficient; return this; } @@ -151,9 +173,11 @@ public PollerOptions build() { return new PollerOptions( maximumPollRateIntervalMilliseconds, maximumPollRatePerSecond, - pollBackoffCoefficient, - pollBackoffInitialInterval, - pollBackoffMaximumInterval, + backoffCoefficient, + backoffInitialInterval, + backoffCongestionInitialInterval, + backoffMaximumInterval, + backoffMaximumJitterCoefficient, pollThreadCount, uncaughtExceptionHandler, pollThreadNamePrefix); @@ -164,9 +188,11 @@ public PollerOptions build() { private final int maximumPollRateIntervalMilliseconds; private final double maximumPollRatePerSecond; - private final double pollBackoffCoefficient; - private final Duration pollBackoffInitialInterval; - private final Duration pollBackoffMaximumInterval; + private final double backoffCoefficient; + private final double backoffMaximumJitterCoefficient; + private final Duration backoffInitialInterval; + private final Duration backoffCongestionInitialInterval; + private final Duration backoffMaximumInterval; private final int pollThreadCount; private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler; private final String pollThreadNamePrefix; @@ -174,17 +200,21 @@ public PollerOptions build() { private PollerOptions( int maximumPollRateIntervalMilliseconds, double maximumPollRatePerSecond, - double pollBackoffCoefficient, - Duration pollBackoffInitialInterval, - Duration pollBackoffMaximumInterval, + double backoffCoefficient, + Duration backoffInitialInterval, + Duration backoffCongestionInitialInterval, + Duration backoffMaximumInterval, + double backoffMaximumJitterCoefficient, int pollThreadCount, Thread.UncaughtExceptionHandler uncaughtExceptionHandler, String pollThreadNamePrefix) { this.maximumPollRateIntervalMilliseconds = maximumPollRateIntervalMilliseconds; this.maximumPollRatePerSecond = maximumPollRatePerSecond; - this.pollBackoffCoefficient = pollBackoffCoefficient; - this.pollBackoffInitialInterval = pollBackoffInitialInterval; - this.pollBackoffMaximumInterval = pollBackoffMaximumInterval; + this.backoffCoefficient = backoffCoefficient; + this.backoffInitialInterval = backoffInitialInterval; + this.backoffCongestionInitialInterval = backoffCongestionInitialInterval; + this.backoffMaximumInterval = backoffMaximumInterval; + this.backoffMaximumJitterCoefficient = backoffMaximumJitterCoefficient; this.pollThreadCount = pollThreadCount; this.uncaughtExceptionHandler = uncaughtExceptionHandler; this.pollThreadNamePrefix = pollThreadNamePrefix; @@ -198,16 +228,24 @@ public double getMaximumPollRatePerSecond() { return maximumPollRatePerSecond; } - public double getPollBackoffCoefficient() { - return pollBackoffCoefficient; + public double getBackoffCoefficient() { + return backoffCoefficient; + } + + public Duration getBackoffInitialInterval() { + return backoffInitialInterval; + } + + public Duration getBackoffCongestionInitialInterval() { + return backoffCongestionInitialInterval; } - public Duration getPollBackoffInitialInterval() { - return pollBackoffInitialInterval; + public Duration getBackoffMaximumInterval() { + return backoffMaximumInterval; } - public Duration getPollBackoffMaximumInterval() { - return pollBackoffMaximumInterval; + public double getBackoffMaximumJitterCoefficient() { + return backoffMaximumJitterCoefficient; } public int getPollThreadCount() { @@ -229,12 +267,16 @@ public String toString() { + maximumPollRateIntervalMilliseconds + ", maximumPollRatePerSecond=" + maximumPollRatePerSecond - + ", pollBackoffCoefficient=" - + pollBackoffCoefficient - + ", pollBackoffInitialInterval=" - + pollBackoffInitialInterval - + ", pollBackoffMaximumInterval=" - + pollBackoffMaximumInterval + + ", backoffCoefficient=" + + backoffCoefficient + + ", backoffInitialInterval=" + + backoffInitialInterval + + ", backoffCongestionInitialInterval=" + + backoffCongestionInitialInterval + + ", backoffMaximumInterval=" + + backoffMaximumInterval + + ", backoffMaximumJitterCoefficient=" + + backoffMaximumJitterCoefficient + ", pollThreadCount=" + pollThreadCount + ", pollThreadNamePrefix='" diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java index 7b0633fd8e..38c9c23143 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java @@ -20,25 +20,34 @@ package io.temporal.internal; +import io.grpc.Status; +import io.grpc.Status.Code; import java.time.Duration; import java.util.Objects; import javax.annotation.Nullable; import javax.annotation.concurrent.NotThreadSafe; /** - * Used to throttle code execution in presence of failures using exponential backoff logic. The - * formula used to calculate the next sleep interval is: + * Used to throttle code execution in presence of failures using exponential backoff logic. + * + *

The formula used to calculate the next sleep interval is: * *

* * 

- * min(pow(backoffCoefficient, failureCount - 1) * initialSleep, maxSleep);
+ * jitter = random number in the range [-maxJitterCoefficient, +maxJitterCoefficient];
+ * sleepTime = min(pow(backoffCoefficient, failureCount - 1) * initialSleep * (1 + jitter), maxSleep);
  *
* + * where initialSleep is either set to regularInitialSleep or + * congestionInitialSleep based on the most recent failure. Note that it means that + * attempt X can possibly get a shorter throttle than attempt X-1, if a non-congestion failure + * occurs after a congestion failure. This is the expected behaviour for all SDK. + * *

Example usage: * * 


- * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
+ * BackoffThrottler throttler = new BackoffThrottler(50, 1000, 60000, 2, 0.1);
  * while(!stopped) {
  *     try {
  *         long throttleMs = throttler.getSleepTime();
@@ -50,7 +59,10 @@
  *         throttler.success();
  *     }
  *     catch (Exception e) {
- *         throttler.failure();
+ *         throttler.failure(
+ *             (e instanceof StatusRuntimeException)
+ *                 ? ((StatusRuntimeException) e).getStatus().getCode()
+ *                 : Status.Code.UNKNOWN);
  *     }
  * }
  *
@@ -60,32 +72,63 @@ @NotThreadSafe public final class BackoffThrottler { - private final Duration initialSleep; + private final Duration regularInitialSleep; + + private final Duration congestionInitialSleep; private final Duration maxSleep; private final double backoffCoefficient; + private final double maxJitterCoefficient; + private int failureCount = 0; + private Status.Code lastFailureCode = Code.OK; + /** * Construct an instance of the throttler. * - * @param initialSleep time to sleep on the first failure + * @param regularInitialSleep time to sleep on the first failure (assuming regular failures) + * @param congestionInitialSleep time to sleep on the first failure (for congestion failures) * @param maxSleep maximum time to sleep independently of number of failures - * @param backoffCoefficient coefficient used to calculate the next time to sleep. + * @param backoffCoefficient coefficient used to calculate the next time to sleep + * @param maxJitterCoefficient maximum jitter coefficient (in the range [0.0, 1.0[) to randomly + * add or subtract to sleep time */ public BackoffThrottler( - Duration initialSleep, @Nullable Duration maxSleep, double backoffCoefficient) { - Objects.requireNonNull(initialSleep, "initialSleep"); - this.initialSleep = initialSleep; + Duration regularInitialSleep, + Duration congestionInitialSleep, + @Nullable Duration maxSleep, + double backoffCoefficient, + double maxJitterCoefficient) { + Objects.requireNonNull(regularInitialSleep, "regularInitialSleep"); + Objects.requireNonNull(congestionInitialSleep, "congestionInitialSleep"); + if (backoffCoefficient < 1.0) { + throw new IllegalArgumentException( + "backoff coefficient less than 1.0: " + backoffCoefficient); + } + if (maxJitterCoefficient < 0 || maxJitterCoefficient >= 1.0) { + throw new IllegalArgumentException( + "maxJitterCoefficient has to be >= 0 and < 1.0: " + maxJitterCoefficient); + } + this.regularInitialSleep = regularInitialSleep; + this.congestionInitialSleep = congestionInitialSleep; this.maxSleep = maxSleep; this.backoffCoefficient = backoffCoefficient; + this.maxJitterCoefficient = maxJitterCoefficient; } public long getSleepTime() { if (failureCount == 0) return 0; - double sleepMillis = Math.pow(backoffCoefficient, failureCount - 1) * initialSleep.toMillis(); + Duration initial = + (lastFailureCode == Code.RESOURCE_EXHAUSTED) ? congestionInitialSleep : regularInitialSleep; + + // Choose a random number in the range [-maxJitterCoefficient, +maxJitterCoefficient]; + double jitter = Math.random() * maxJitterCoefficient * 2 - maxJitterCoefficient; + double sleepMillis = + Math.pow(backoffCoefficient, failureCount - 1) * initial.toMillis() * (1 + jitter); + if (maxSleep != null) { return Math.min((long) sleepMillis, maxSleep.toMillis()); } @@ -96,13 +139,15 @@ public int getAttemptCount() { return failureCount; } - /** Reset failure count to 0. */ + /** Reset failure count to 0 and clear last failure code. */ public void success() { failureCount = 0; + lastFailureCode = Code.OK; } - /** Increment failure count. */ - public void failure() { + /** Increment failure count and set last failure code. */ + public void failure(Status.Code failureCode) { failureCount++; + lastFailureCode = failureCode; } } diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java index ee5ebbe54d..65ebab1d45 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java @@ -22,6 +22,7 @@ import io.grpc.Context; import io.grpc.Deadline; +import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; import io.temporal.internal.BackoffThrottler; @@ -66,8 +67,10 @@ public GrpcAsyncRetryer( this.throttler = new BackoffThrottler( rpcOptions.getInitialInterval(), + rpcOptions.getCongestionInitialInterval(), rpcOptions.getMaximumInterval(), - rpcOptions.getBackoffCoefficient()); + rpcOptions.getBackoffCoefficient(), + rpcOptions.getMaximumJitterCoefficient()); } public CompletableFuture retry() { @@ -103,13 +106,19 @@ private void retry(CompletableFuture resultCF) { throttler.success(); resultCF.complete(r); } else { - throttler.failure(); + throttler.failure( + (e instanceof StatusRuntimeException) + ? ((StatusRuntimeException) e).getStatus().getCode() + : Status.Code.UNKNOWN); failOrRetry(e, resultCF); } }); } catch (Throwable e) { - throttler.failure(); + throttler.failure( + (e instanceof StatusRuntimeException) + ? ((StatusRuntimeException) e).getStatus().getCode() + : Status.Code.UNKNOWN); // function isn't supposed to throw exceptions, it should always return a // CompletableFuture even if it's a failed one. // But if this happens - process the same way as it would be an exception from diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java index e80444a3b2..2202bb39a4 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java @@ -48,8 +48,10 @@ public R retry( BackoffThrottler throttler = new BackoffThrottler( rpcOptions.getInitialInterval(), + rpcOptions.getCongestionInitialInterval(), rpcOptions.getMaximumInterval(), - rpcOptions.getBackoffCoefficient()); + rpcOptions.getBackoffCoefficient(), + rpcOptions.getMaximumJitterCoefficient()); int attempt = 0; StatusRuntimeException lastMeaningfulException = null; @@ -79,7 +81,7 @@ public R retry( } lastMeaningfulException = GrpcRetryerUtils.lastMeaningfulException(e, lastMeaningfulException); - throttler.failure(); + throttler.failure(e.getStatus().getCode()); } // No catch block for any other exceptions because we don't retry them, we pass them through. // It's designed this way because it's GrpcRetryer, not general purpose retryer. diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java index ae87c1caea..f09be6b823 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java @@ -85,6 +85,8 @@ public static final class Builder { private Duration initialInterval; + private Duration congestionInitialInterval; + private Duration expiration; private double backoffCoefficient; @@ -93,6 +95,10 @@ public static final class Builder { private Duration maximumInterval; + // 0 is a valid value for maximumJitterCoefficient, and yet, it is not the default value for + // maximumJitterCoefficient. Thus, use -1 instead as a marker for "use default value" + private double maximumJitterCoefficient = -1.0; + private List doNotRetry = new ArrayList<>(); private Builder() {} @@ -105,27 +111,58 @@ private Builder(RpcRetryOptions options) { this.maximumAttempts = options.getMaximumAttempts(); this.expiration = options.getExpiration(); this.initialInterval = options.getInitialInterval(); + this.congestionInitialInterval = options.getCongestionInitialInterval(); this.maximumInterval = options.getMaximumInterval(); + this.maximumJitterCoefficient = options.getMaximumJitterCoefficient(); this.doNotRetry = options.getDoNotRetry(); } /** - * Interval of the first retry. If coefficient is 1.0 then it is used for all retries. Required. + * Interval of the first retry, on regular failures. If coefficient is 1.0 then it is used for + * all retries. Defaults to 50ms. + * + * @param initialInterval Interval to wait on first retry. Default will be used if set to {@code + * null}. */ public Builder setInitialInterval(Duration initialInterval) { - Objects.requireNonNull(initialInterval); - if (initialInterval.isNegative() || initialInterval.isZero()) { - throw new IllegalArgumentException("Invalid interval: " + initialInterval); + if (initialInterval != null) { + if (initialInterval.isNegative() || initialInterval.isZero()) { + throw new IllegalArgumentException("Invalid interval: " + initialInterval); + } + this.initialInterval = initialInterval; + } else { + this.initialInterval = null; + } + return this; + } + + /** + * Interval of the first retry, on congestion related failures (i.e. RESOURCE_EXHAUSTED errors). + * If coefficient is 1.0 then it is used for all retries. Defaults to 1000ms. + * + * @param congestionInitialInterval Interval to wait on first retry, on congestion failures. + * Defaults to 1000ms, which is used if set to {@code null}. + */ + public Builder setCongestionInitialInterval(Duration congestionInitialInterval) { + if (initialInterval != null) { + if (congestionInitialInterval.isNegative() || congestionInitialInterval.isZero()) { + throw new IllegalArgumentException("Invalid interval: " + congestionInitialInterval); + } + this.congestionInitialInterval = congestionInitialInterval; + } else { + this.congestionInitialInterval = null; } - this.initialInterval = initialInterval; return this; } /** * Maximum time to retry. When exceeded the retries stop even if maximum retries is not reached - * yet. + * yet. Defaults to 1 minute. * - *

At least one of the expiration or {@link #setMaximumAttempts(int)} is required to be set. + *

At least one of expiration or {@link #setMaximumAttempts(int)} is required to be set. + * + * @param expiration Maximum time to retry. Defaults to 1 minute, which is used if set to {@code + * null}. */ public Builder setExpiration(Duration expiration) { if (expiration != null) { @@ -142,12 +179,19 @@ public Builder setExpiration(Duration expiration) { /** * Coefficient used to calculate the next retry interval. The next retry interval is previous * interval multiplied by this coefficient. Must be 1 or larger. Default is 2.0. + * + * @param backoffCoefficient Coefficient used to calculate the next retry interval. Defaults to + * 2.0, which is used if set to {@code 0}. */ public Builder setBackoffCoefficient(double backoffCoefficient) { - if (backoffCoefficient < 1d) { - throw new IllegalArgumentException("coefficient less than 1.0: " + backoffCoefficient); + if (backoffCoefficient != 0.0) { + if (!Double.isFinite(backoffCoefficient) || (backoffCoefficient < 1.0)) { + throw new IllegalArgumentException("coefficient has to be >= 1.0: " + backoffCoefficient); + } + this.backoffCoefficient = backoffCoefficient; + } else { + this.backoffCoefficient = 0.0; } - this.backoffCoefficient = backoffCoefficient; return this; } @@ -155,10 +199,11 @@ public Builder setBackoffCoefficient(double backoffCoefficient) { * When exceeded the amount of attempts, stop. Even if expiration time is not reached.
* Default is unlimited. * - *

At least one of the maximum attempts or {@link #setExpiration(Duration)} is required to be + *

At least one of maximum attempts or {@link #setExpiration(Duration)} is required to be * set. * - * @param maximumAttempts Maximum number of attempts. Default will be used if set to {@code 0}. + * @param maximumAttempts Maximum number of attempts. Defaults to unlimited, which is used if + * set to {@code 0}. */ public Builder setMaximumAttempts(int maximumAttempts) { if (maximumAttempts < 0) { @@ -173,14 +218,39 @@ public Builder setMaximumAttempts(int maximumAttempts) { * is the cap of the increase.
* Default is 100x of initial interval. Can't be less than {@link #setInitialInterval(Duration)} * - * @param maximumInterval the maximum interval value. Default will be used if set to {@code - * null}. + * @param maximumInterval the maximum interval value. Defaults to 100x initial interval, which + * is used if set to {@code null}. */ public Builder setMaximumInterval(Duration maximumInterval) { - if (maximumInterval != null && (maximumInterval.isNegative() || maximumInterval.isZero())) { - throw new IllegalArgumentException("Invalid interval: " + maximumInterval); + if (maximumInterval != null) { + if ((maximumInterval.isNegative() || maximumInterval.isZero())) { + throw new IllegalArgumentException("Invalid interval: " + maximumInterval); + } + this.maximumInterval = maximumInterval; + } else { + this.maximumInterval = null; + } + return this; + } + + /** + * Maximum amount of jitter to apply. 0.2 means that actual retry time can be +/- 20% of the + * calculated time. Set to 0 to disable jitter. Must be lower than 1. Default is 0.1. + * + * @param maximumJitterCoefficient Maximum amount of jitter. Default will be used if set to -1. + */ + public Builder setMaximumJitterCoefficient(double maximumJitterCoefficient) { + if (maximumJitterCoefficient != -1.0) { + if (!Double.isFinite(maximumJitterCoefficient) + || maximumJitterCoefficient < 0 + || maximumJitterCoefficient >= 1.0) { + throw new IllegalArgumentException( + "maximumJitterCoefficient has to be >= 0 and < 1.0: " + maximumJitterCoefficient); + } + this.maximumJitterCoefficient = maximumJitterCoefficient; + } else { + this.maximumJitterCoefficient = -1.0; } - this.maximumInterval = maximumInterval; return this; } @@ -244,12 +314,17 @@ public Builder setRetryOptions(RpcRetryOptions o) { } setInitialInterval( OptionsUtils.merge(initialInterval, o.getInitialInterval(), Duration.class)); + setCongestionInitialInterval( + OptionsUtils.merge(congestionInitialInterval, o.getInitialInterval(), Duration.class)); setExpiration(OptionsUtils.merge(expiration, o.getExpiration(), Duration.class)); setMaximumInterval( OptionsUtils.merge(maximumInterval, o.getMaximumInterval(), Duration.class)); setBackoffCoefficient( OptionsUtils.merge(backoffCoefficient, o.getBackoffCoefficient(), double.class)); setMaximumAttempts(OptionsUtils.merge(maximumAttempts, o.getMaximumAttempts(), int.class)); + if (o.getMaximumJitterCoefficient() != -1.0) { + setMaximumJitterCoefficient(o.getMaximumJitterCoefficient()); + } doNotRetry = merge(doNotRetry, o.getDoNotRetry()); validateBuildWithDefaults(); return this; @@ -272,10 +347,12 @@ private List merge(List o1, List public RpcRetryOptions build() { return new RpcRetryOptions( initialInterval, + congestionInitialInterval, backoffCoefficient, expiration, maximumAttempts, maximumInterval, + maximumJitterCoefficient, doNotRetry); } @@ -294,6 +371,12 @@ public RpcRetryOptions validateBuildWithDefaults() { if (initialInterval == null || initialInterval.isZero() || initialInterval.isNegative()) { initialInterval = DefaultStubServiceOperationRpcRetryOptions.INITIAL_INTERVAL; } + if (congestionInitialInterval == null + || congestionInitialInterval.isZero() + || congestionInitialInterval.isNegative()) { + congestionInitialInterval = + DefaultStubServiceOperationRpcRetryOptions.CONGESTION_INITIAL_INTERVAL; + } if (expiration == null || expiration.isZero() || expiration.isNegative()) { expiration = DefaultStubServiceOperationRpcRetryOptions.EXPIRATION_INTERVAL; } @@ -308,13 +391,20 @@ public RpcRetryOptions validateBuildWithDefaults() { } } + if (maximumJitterCoefficient == -1.0) { + maximumJitterCoefficient = + DefaultStubServiceOperationRpcRetryOptions.MAXIMUM_JITTER_COEFFICIENT; + } + RpcRetryOptions result = new RpcRetryOptions( initialInterval, + congestionInitialInterval, backoff, expiration, maximumAttempts, maxInterval, + maximumJitterCoefficient, MoreObjects.firstNonNull(doNotRetry, Collections.emptyList())); result.validate(); return result; @@ -323,6 +413,8 @@ public RpcRetryOptions validateBuildWithDefaults() { private final Duration initialInterval; + private final Duration congestionInitialInterval; + private final double backoffCoefficient; private final Duration expiration; @@ -331,20 +423,26 @@ public RpcRetryOptions validateBuildWithDefaults() { private final Duration maximumInterval; + private final double maximumJitterCoefficient; + private final @Nonnull List doNotRetry; private RpcRetryOptions( Duration initialInterval, + Duration congestionInitialInterval, double backoffCoefficient, Duration expiration, int maximumAttempts, Duration maximumInterval, + double maximumJitterCoefficient, @Nonnull List doNotRetry) { this.initialInterval = initialInterval; + this.congestionInitialInterval = congestionInitialInterval; this.backoffCoefficient = backoffCoefficient; this.expiration = expiration; this.maximumAttempts = maximumAttempts; this.maximumInterval = maximumInterval; + this.maximumJitterCoefficient = maximumJitterCoefficient; this.doNotRetry = Collections.unmodifiableList(doNotRetry); } @@ -352,6 +450,10 @@ public Duration getInitialInterval() { return initialInterval; } + public Duration getCongestionInitialInterval() { + return congestionInitialInterval; + } + public double getBackoffCoefficient() { return backoffCoefficient; } @@ -368,6 +470,10 @@ public Duration getMaximumInterval() { return maximumInterval; } + public double getMaximumJitterCoefficient() { + return maximumJitterCoefficient; + } + public void validate() { validate(true); } @@ -376,6 +482,9 @@ public void validate(boolean hasToBeFinite) { if (initialInterval == null) { throw new IllegalStateException("required property initialInterval not set"); } + if (congestionInitialInterval == null) { + throw new IllegalStateException("required property congestionInitialInterval not set"); + } if (maximumAttempts < 0) { throw new IllegalArgumentException("negative maximum attempts"); } @@ -393,6 +502,12 @@ public void validate(boolean hasToBeFinite) { if (backoffCoefficient != 0d && backoffCoefficient < 1.0) { throw new IllegalArgumentException("coefficient less than 1"); } + if (!Double.isFinite(maximumJitterCoefficient) + || maximumJitterCoefficient < 0 + || maximumJitterCoefficient >= 1.0) { + throw new IllegalArgumentException( + "maximumJitterCoefficient has to be >= 0 and < 1.0: " + maximumJitterCoefficient); + } } public @Nonnull List getDoNotRetry() { @@ -404,6 +519,8 @@ public String toString() { return "RetryOptions{" + "initialInterval=" + initialInterval + + "congestionInitialInterval=" + + congestionInitialInterval + ", backoffCoefficient=" + backoffCoefficient + ", expiration=" @@ -412,6 +529,8 @@ public String toString() { + maximumAttempts + ", maximumInterval=" + maximumInterval + + ", maximumJitterCoefficient=" + + maximumJitterCoefficient + ", doNotRetry=" + doNotRetry + '}'; @@ -425,8 +544,10 @@ public boolean equals(Object o) { return Double.compare(that.backoffCoefficient, backoffCoefficient) == 0 && maximumAttempts == that.maximumAttempts && Objects.equals(initialInterval, that.initialInterval) + && Objects.equals(congestionInitialInterval, that.congestionInitialInterval) && Objects.equals(expiration, that.expiration) && Objects.equals(maximumInterval, that.maximumInterval) + && Objects.equals(maximumJitterCoefficient, that.maximumJitterCoefficient) && Objects.equals(doNotRetry, that.doNotRetry); } @@ -434,10 +555,12 @@ public boolean equals(Object o) { public int hashCode() { return Objects.hash( initialInterval, + congestionInitialInterval, backoffCoefficient, expiration, maximumAttempts, maximumInterval, + maximumJitterCoefficient, doNotRetry); } } diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubLongPollRpcRetryOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubLongPollRpcRetryOptions.java index f291c63bca..011fc7b6e7 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubLongPollRpcRetryOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubLongPollRpcRetryOptions.java @@ -26,8 +26,10 @@ /** Default rpc retry options for long polls like waiting for the workflow finishing and result. */ public class DefaultStubLongPollRpcRetryOptions { public static final Duration INITIAL_INTERVAL = Duration.ofMillis(50); + public static final Duration CONGESTION_INITIAL_INTERVAL = Duration.ofMillis(1000); public static final Duration MAXIMUM_INTERVAL = Duration.ofMinutes(1); public static final double BACKOFF = 1.2; + public static final double MAXIMUM_JITTER_COEFFICIENT = 0.1; // partial build because expiration is not set, long polls work with absolute deadlines instead public static final RpcRetryOptions INSTANCE = getBuilder().build(); @@ -42,8 +44,10 @@ private static RpcRetryOptions.Builder getBuilder() { RpcRetryOptions.Builder roBuilder = RpcRetryOptions.newBuilder() .setInitialInterval(INITIAL_INTERVAL) + .setCongestionInitialInterval(CONGESTION_INITIAL_INTERVAL) .setBackoffCoefficient(BACKOFF) - .setMaximumInterval(MAXIMUM_INTERVAL); + .setMaximumInterval(MAXIMUM_INTERVAL) + .setMaximumJitterCoefficient(MAXIMUM_JITTER_COEFFICIENT); return roBuilder; } diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubServiceOperationRpcRetryOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubServiceOperationRpcRetryOptions.java index 4123235ef7..2683c7b9e1 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubServiceOperationRpcRetryOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/rpcretry/DefaultStubServiceOperationRpcRetryOptions.java @@ -30,9 +30,11 @@ */ public class DefaultStubServiceOperationRpcRetryOptions { public static final Duration INITIAL_INTERVAL = Duration.ofMillis(50); + public static final Duration CONGESTION_INITIAL_INTERVAL = Duration.ofMillis(1000); public static final Duration EXPIRATION_INTERVAL = Duration.ofMinutes(1); public static final Duration MAXIMUM_INTERVAL; public static final double BACKOFF = 2; + public static final double MAXIMUM_JITTER_COEFFICIENT = 0.1; public static final RpcRetryOptions INSTANCE; @@ -49,8 +51,10 @@ public class DefaultStubServiceOperationRpcRetryOptions { public static RpcRetryOptions.Builder getBuilder() { return RpcRetryOptions.newBuilder() .setInitialInterval(INITIAL_INTERVAL) + .setCongestionInitialInterval(CONGESTION_INITIAL_INTERVAL) .setExpiration(EXPIRATION_INTERVAL) .setBackoffCoefficient(BACKOFF) - .setMaximumInterval(MAXIMUM_INTERVAL); + .setMaximumInterval(MAXIMUM_INTERVAL) + .setMaximumJitterCoefficient(MAXIMUM_JITTER_COEFFICIENT); } } diff --git a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java index ffe43c8d9a..5732081a29 100644 --- a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java +++ b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java @@ -58,6 +58,7 @@ public void testExpirationAsync() throws InterruptedException { .setInitialInterval(Duration.ofMillis(10)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -79,7 +80,7 @@ public void testExpirationAsync() throws InterruptedException { } assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1); - assertTrue(System.currentTimeMillis() - start > 500); + assertTrue(System.currentTimeMillis() - start >= 500); } @Test @@ -91,6 +92,7 @@ public void testExpirationFutureAsync() throws InterruptedException { .setInitialInterval(Duration.ofMillis(10)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -115,7 +117,7 @@ public void testExpirationFutureAsync() throws InterruptedException { } assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1); - assertTrue(System.currentTimeMillis() - start > 500); + assertTrue(System.currentTimeMillis() - start >= 500); } @Test @@ -126,6 +128,7 @@ public void testDoNotRetryAsync() throws InterruptedException { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) .addDoNotRetry(STATUS_CODE, null) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); @@ -161,6 +164,7 @@ public void testInterruptedExceptionAsync() throws InterruptedException { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -193,6 +197,7 @@ public void testNotStatusRuntimeExceptionAsync() throws InterruptedException { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -227,6 +232,7 @@ public void testRetryDeadlineExceededException() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -259,7 +265,7 @@ public void testRetryDeadlineExceededException() { assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.", - System.currentTimeMillis() - start > 500); + System.currentTimeMillis() - start >= 500); assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.", @@ -273,6 +279,7 @@ public void testRespectGlobalDeadlineExceeded() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(50000)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -321,6 +328,7 @@ public void testGlobalDeadlineExceededAfterAnotherException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); final AtomicReference exception = new AtomicReference<>(); @@ -359,4 +367,46 @@ public void testGlobalDeadlineExceededAfterAnotherException() { Status.Code.DATA_LOSS, exception.get().getStatus().getCode()); } + + @Test + public void testResourceExhaustedFailure() throws InterruptedException { + RpcRetryOptions options = + RpcRetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(1)) + .setCongestionInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) + .setMaximumAttempts(3) + .validateBuildWithDefaults(); + long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); + + try { + new GrpcAsyncRetryer<>( + scheduledExecutor, + () -> { + attempts.incrementAndGet(); + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally( + new StatusRuntimeException(Status.fromCode(Status.Code.RESOURCE_EXHAUSTED))); + return future; + }, + new GrpcRetryer.GrpcRetryerOptions(options, null), + GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() + .get(); + fail("unreachable"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof StatusRuntimeException); + assertEquals( + Status.Code.RESOURCE_EXHAUSTED, + ((StatusRuntimeException) e.getCause()).getStatus().getCode()); + } + + long elapsedTime = System.currentTimeMillis() - start; + assertTrue("We should retry RESOURCE_EXHAUSTED failures.", attempts.get() > 2); + assertTrue( + "We should retry RESOURCE_EXHAUSTED failures using congestionInitialInterval.", + elapsedTime >= 2000); + } } diff --git a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java index 3124dd7c5b..b1ecdaff70 100644 --- a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java +++ b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java @@ -63,6 +63,7 @@ public void testExpiration() { .setInitialInterval(Duration.ofMillis(10)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -81,7 +82,7 @@ public void testExpiration() { } assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1); - assertTrue(System.currentTimeMillis() - start > 500); + assertTrue(System.currentTimeMillis() - start >= 500); } @Test @@ -92,6 +93,7 @@ public void testDoNotRetry() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) .addDoNotRetry(STATUS_CODE, null) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); @@ -121,6 +123,7 @@ public void testInterruptedException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -146,6 +149,7 @@ public void testNotStatusRuntimeException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -175,6 +179,7 @@ public void testRetryDeadlineExceededException() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -195,7 +200,7 @@ public void testRetryDeadlineExceededException() { assertEquals(Status.Code.DEADLINE_EXCEEDED, e.getStatus().getCode()); assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.", - System.currentTimeMillis() - start > 500); + System.currentTimeMillis() - start >= 500); assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.", @@ -209,6 +214,7 @@ public void testRespectGlobalDeadlineExceeded() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(50000)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -246,6 +252,7 @@ public void testGlobalDeadlineExceededAfterAnotherException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); final AtomicReference exception = new AtomicReference<>(); @@ -276,4 +283,59 @@ public void testGlobalDeadlineExceededAfterAnotherException() { Status.Code.DATA_LOSS, exception.get().getStatus().getCode()); } + + @Test + public void testResourceExhaustedFailure() { + RpcRetryOptions options = + RpcRetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(1)) + .setCongestionInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitterCoefficient(0) + .setMaximumAttempts(3) + .validateBuildWithDefaults(); + long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); + + StatusRuntimeException e = + assertThrows( + StatusRuntimeException.class, + () -> + DEFAULT_SYNC_RETRYER.retry( + () -> { + attempts.incrementAndGet(); + throw new StatusRuntimeException( + Status.fromCode(Status.Code.RESOURCE_EXHAUSTED)); + }, + new GrpcRetryer.GrpcRetryerOptions(options, null), + GetSystemInfoResponse.Capabilities.getDefaultInstance())); + + assertEquals(Status.Code.RESOURCE_EXHAUSTED, e.getStatus().getCode()); + long elapsedTime = System.currentTimeMillis() - start; + assertTrue("We should retry RESOURCE_EXHAUSTED failures.", attempts.get() > 2); + assertTrue( + "We should retry RESOURCE_EXHAUSTED failures using congestionInitialInterval.", + elapsedTime >= 2000); + } + + @Test + public void testCongestionAndJitterAreNotMandatory() { + RpcRetryOptions options = + RpcRetryOptions.newBuilder( + RpcRetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(1)) + .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumAttempts(3) + .build()) + .validateBuildWithDefaults(); + + // The following options matches the values above + assertEquals(Duration.ofMillis(1), options.getInitialInterval()); + assertEquals(Duration.ofMillis(1000), options.getMaximumInterval()); + assertEquals(3, options.getMaximumAttempts()); + + // The following were added latter; they must silently use default values if unspecified + assertEquals(Duration.ofMillis(1000), options.getCongestionInitialInterval()); + assertEquals(0.1, options.getMaximumJitterCoefficient(), 0.01); + } }