From c46d980ca9c50dfd908ff41e6ee7aa944885d195 Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Thu, 13 Oct 2022 17:03:17 -0400 Subject: [PATCH 1/6] Use longer retry interval on RESOURCE_EXHAUSTED --- .../io/temporal/internal/worker/Poller.java | 13 +- .../internal/worker/PollerOptions.java | 118 ++++++++++++------ .../temporal/internal/BackoffThrottler.java | 59 +++++++-- .../internal/retryer/GrpcAsyncRetryer.java | 15 ++- .../internal/retryer/GrpcRetryerUtils.java | 2 + .../internal/retryer/GrpcSyncRetryer.java | 6 +- .../serviceclient/RpcRetryOptions.java | 87 ++++++++++++- .../DefaultStubLongPollRpcRetryOptions.java | 6 +- ...ltStubServiceOperationRpcRetryOptions.java | 6 +- .../retryer/GrpcAsyncRetryerTest.java | 50 ++++++++ .../internal/retryer/GrpcSyncRetryerTest.java | 41 ++++++ 11 files changed, 340 insertions(+), 63 deletions(-) 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..7adf6b7a57 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.getBackoffMaximumJitter()); } @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..dd143783d8 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 backoffMaximumJitter = 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.backoffMaximumJitter = options.getBackoffMaximumJitter(); 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 setBackoffMaximumJitter(double backoffMaximumJitter) { + this.backoffMaximumJitter = backoffMaximumJitter; return this; } @@ -151,9 +173,11 @@ public PollerOptions build() { return new PollerOptions( maximumPollRateIntervalMilliseconds, maximumPollRatePerSecond, - pollBackoffCoefficient, - pollBackoffInitialInterval, - pollBackoffMaximumInterval, + backoffCoefficient, + backoffInitialInterval, + backoffCongestionInitialInterval, + backoffMaximumInterval, + backoffMaximumJitter, 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 backoffMaximumJitter; + 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 backoffMaximumJitter, 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.backoffMaximumJitter = backoffMaximumJitter; 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 getBackoffMaximumJitter() { + return backoffMaximumJitter; } 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 + + ", backoffMaximumJitter=" + + backoffMaximumJitter + ", 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 2e59513055..2ed1c4afb6 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java @@ -20,6 +20,8 @@ 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; @@ -32,7 +34,7 @@ *

* * 

- * min(pow(backoffCoefficient, failureCount - 1) * initialSleep, maxSleep);
+ * min(pow(backoffCoefficient, failureCount - 1) * initialSleep * (1 + jitter), maxSleep);
  *
* *

Example usage: @@ -40,7 +42,7 @@ *

* * 

- * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
+ * BackoffThrottler throttler = new BackoffThrottler(50, 1000, 60000, 2, 0.1);
  * while(!stopped) {
  *     try {
  *         long throttleMs = throttler.getSleepTime();
@@ -52,7 +54,10 @@
  *         throttler.success();
  *     }
  *     catch (Exception e) {
- *         throttler.failure();
+ *         throttler.failure(
+ *             (e instanceof StatusRuntimeException)
+ *                 ? ((StatusRuntimeException) e).getStatus().getCode()
+ *                 : Status.Code.UNKNOWN);
  *     }
  * }
  *
@@ -62,32 +67,58 @@ @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 maxJitter; + 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 maxJitter maximum jitter to add or subtract */ 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 maxJitter) { + Objects.requireNonNull(regularInitialSleep, "regularInitialSleep"); + Objects.requireNonNull(congestionInitialSleep, "congestionInitialSleep"); + if (backoffCoefficient < 1.0) { + throw new IllegalArgumentException( + "backoff coefficient less than 1.0: " + backoffCoefficient); + } + if (maxJitter < 0 || maxJitter >= 1.0) { + throw new IllegalArgumentException("maximumJitter has to be >= 0 and < 1.0: " + maxJitter); + } + this.regularInitialSleep = regularInitialSleep; + this.congestionInitialSleep = congestionInitialSleep; this.maxSleep = maxSleep; this.backoffCoefficient = backoffCoefficient; + this.maxJitter = maxJitter; } 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; + double jitter = Math.random() * maxJitter * 2 - maxJitter; + double sleepMillis = + Math.pow(backoffCoefficient, failureCount - 1) * initial.toMillis() * (1 + jitter); if (maxSleep != null) { return Math.min((long) sleepMillis, maxSleep.toMillis()); } @@ -98,13 +129,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..fbdfb5cdef 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.getMaximumJitter()); } 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/GrpcRetryerUtils.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryerUtils.java index 335bfb97aa..2f9d993723 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryerUtils.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryerUtils.java @@ -76,6 +76,8 @@ class GrpcRetryerUtils { // By default, we keep retrying with DEADLINE_EXCEEDED assuming that it's the deadline of // one attempt which expired, but not the whole sequence. break; + case RESOURCE_EXHAUSTED: + break; default: for (RpcRetryOptions.DoNotRetryItem pair : options.getDoNotRetry()) { if (pair.getCode() == code 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..e4ff29cf33 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.getMaximumJitter()); 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..f75b156c9e 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 maximumJitter, and yet, it is not the default value for maximumJitter. + // Use NaN instead as a marker for unconfigured jitter + private double maximumJitter = Double.NaN; + private List doNotRetry = new ArrayList<>(); private Builder() {} @@ -105,12 +111,15 @@ 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.maximumJitter = options.getMaximumJitter(); 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. Required. */ public Builder setInitialInterval(Duration initialInterval) { Objects.requireNonNull(initialInterval); @@ -121,6 +130,19 @@ public Builder setInitialInterval(Duration initialInterval) { 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. + */ + public Builder setCongestionInitialInterval(Duration congestionInitialInterval) { + Objects.requireNonNull(congestionInitialInterval); + if (congestionInitialInterval.isNegative() || congestionInitialInterval.isZero()) { + throw new IllegalArgumentException("Invalid interval: " + congestionInitialInterval); + } + this.congestionInitialInterval = congestionInitialInterval; + return this; + } + /** * Maximum time to retry. When exceeded the retries stop even if maximum retries is not reached * yet. @@ -184,6 +206,19 @@ public Builder setMaximumInterval(Duration maximumInterval) { 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 setMaximumJitter(double maximumJitter) { + if (!Double.isFinite(maximumJitter) || maximumJitter < 0 || maximumJitter >= 1.0) { + throw new IllegalArgumentException( + "maximumJitter has to be >= 0 and < 1.0: " + maximumJitter); + } + this.maximumJitter = maximumJitter; + return this; + } + /** * Makes request that receives a server response with gRPC {@code code} and failure of {@code * detailsClass} type non-retryable. @@ -244,12 +279,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 (Double.isFinite(o.getMaximumJitter())) { + setMaximumJitter(o.getMaximumJitter()); + } doNotRetry = merge(doNotRetry, o.getDoNotRetry()); validateBuildWithDefaults(); return this; @@ -272,10 +312,12 @@ private List merge(List o1, List public RpcRetryOptions build() { return new RpcRetryOptions( initialInterval, + congestionInitialInterval, backoffCoefficient, expiration, maximumAttempts, maximumInterval, + maximumJitter, doNotRetry); } @@ -294,6 +336,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 +356,19 @@ public RpcRetryOptions validateBuildWithDefaults() { } } + if (Double.isNaN(maximumJitter)) { + maximumJitter = DefaultStubServiceOperationRpcRetryOptions.MAXIMUM_JITTER; + } + RpcRetryOptions result = new RpcRetryOptions( initialInterval, + congestionInitialInterval, backoff, expiration, maximumAttempts, maxInterval, + maximumJitter, MoreObjects.firstNonNull(doNotRetry, Collections.emptyList())); result.validate(); return result; @@ -323,6 +377,8 @@ public RpcRetryOptions validateBuildWithDefaults() { private final Duration initialInterval; + private final Duration congestionInitialInterval; + private final double backoffCoefficient; private final Duration expiration; @@ -331,20 +387,26 @@ public RpcRetryOptions validateBuildWithDefaults() { private final Duration maximumInterval; + private final double maximumJitter; + private final @Nonnull List doNotRetry; private RpcRetryOptions( Duration initialInterval, + Duration congestionInitialInterval, double backoffCoefficient, Duration expiration, int maximumAttempts, Duration maximumInterval, + double maximumJitter, @Nonnull List doNotRetry) { this.initialInterval = initialInterval; + this.congestionInitialInterval = congestionInitialInterval; this.backoffCoefficient = backoffCoefficient; this.expiration = expiration; this.maximumAttempts = maximumAttempts; this.maximumInterval = maximumInterval; + this.maximumJitter = maximumJitter; this.doNotRetry = Collections.unmodifiableList(doNotRetry); } @@ -352,6 +414,10 @@ public Duration getInitialInterval() { return initialInterval; } + public Duration getCongestionInitialInterval() { + return congestionInitialInterval; + } + public double getBackoffCoefficient() { return backoffCoefficient; } @@ -368,6 +434,10 @@ public Duration getMaximumInterval() { return maximumInterval; } + public double getMaximumJitter() { + return maximumJitter; + } + public void validate() { validate(true); } @@ -376,6 +446,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 longInitialInterval not set"); + } if (maximumAttempts < 0) { throw new IllegalArgumentException("negative maximum attempts"); } @@ -393,6 +466,10 @@ public void validate(boolean hasToBeFinite) { if (backoffCoefficient != 0d && backoffCoefficient < 1.0) { throw new IllegalArgumentException("coefficient less than 1"); } + if (!Double.isFinite(maximumJitter) || maximumJitter < 0 || maximumJitter >= 1.0) { + throw new IllegalArgumentException( + "maximumJitter has to be >= 0 and < 1.0: " + maximumJitter); + } } public @Nonnull List getDoNotRetry() { @@ -404,6 +481,8 @@ public String toString() { return "RetryOptions{" + "initialInterval=" + initialInterval + + "congestionInitialInterval=" + + congestionInitialInterval + ", backoffCoefficient=" + backoffCoefficient + ", expiration=" @@ -412,6 +491,8 @@ public String toString() { + maximumAttempts + ", maximumInterval=" + maximumInterval + + ", maximumJitter=" + + maximumJitter + ", doNotRetry=" + doNotRetry + '}'; @@ -425,8 +506,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(maximumJitter, that.maximumJitter) && Objects.equals(doNotRetry, that.doNotRetry); } @@ -434,10 +517,12 @@ public boolean equals(Object o) { public int hashCode() { return Objects.hash( initialInterval, + congestionInitialInterval, backoffCoefficient, expiration, maximumAttempts, maximumInterval, + maximumJitter, 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..ba76db72a3 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 = 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) + .setMaximumJitter(MAXIMUM_JITTER); 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..399ad0803f 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 = 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) + .setMaximumJitter(MAXIMUM_JITTER); } } 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..ec4859fc32 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)) + .setMaximumJitter(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -91,6 +92,7 @@ public void testExpirationFutureAsync() throws InterruptedException { .setInitialInterval(Duration.ofMillis(10)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) + .setMaximumJitter(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -126,6 +128,7 @@ public void testDoNotRetryAsync() throws InterruptedException { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitter(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)) + .setMaximumJitter(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)) + .setMaximumJitter(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)) + .setMaximumJitter(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -273,6 +279,7 @@ public void testRespectGlobalDeadlineExceeded() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(50000)) + .setMaximumJitter(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)) + .setMaximumJitter(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)) + .setMaximumJitter(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 longInitialInterval.", + 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..65e47f332b 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)) + .setMaximumJitter(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -92,6 +93,7 @@ public void testDoNotRetry() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) + .setMaximumJitter(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)) + .setMaximumJitter(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)) + .setMaximumJitter(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)) + .setMaximumJitter(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -209,6 +214,7 @@ public void testRespectGlobalDeadlineExceeded() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(50000)) + .setMaximumJitter(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)) + .setMaximumJitter(0) .validateBuildWithDefaults(); final AtomicReference exception = new AtomicReference<>(); @@ -276,4 +283,38 @@ 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)) + .setMaximumJitter(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 longInitialInterval.", + elapsedTime > 2000); + } } From e686d36480fdf71fce7be73996dd63a3664c30d4 Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Fri, 14 Oct 2022 15:33:04 -0400 Subject: [PATCH 2/6] Test: assert existing RpcRetryOptions in the wild will silently get defaults --- .../serviceclient/RpcRetryOptions.java | 2 +- .../internal/retryer/GrpcSyncRetryerTest.java | 21 +++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) 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 f75b156c9e..e88ee64bdb 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java @@ -96,7 +96,7 @@ public static final class Builder { private Duration maximumInterval; // 0 is a valid value for maximumJitter, and yet, it is not the default value for maximumJitter. - // Use NaN instead as a marker for unconfigured jitter + // Use NaN instead as a marker for not configured jitter private double maximumJitter = Double.NaN; private List doNotRetry = new ArrayList<>(); 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 65e47f332b..1680fe74a7 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 @@ -317,4 +317,25 @@ public void testResourceExhaustedFailure() { "We should retry RESOURCE_EXHAUSTED failures using longInitialInterval.", 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.getMaximumJitter(), 0.01); + } } From 9cb4c506f96e16a03e943bc4e52b4bd04e421b78 Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Fri, 14 Oct 2022 15:59:34 -0400 Subject: [PATCH 3/6] Add a comment addressing a review concern --- .../java/io/temporal/internal/BackoffThrottler.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) 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 2ed1c4afb6..59fff8c88e 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java @@ -28,8 +28,10 @@ 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: * *

* @@ -37,6 +39,10 @@ * 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 bahaviour for all SDK. + * *

Example usage: * *

From 7dec853d0b17cf0edfeb216ff0c4e709eb3936a9 Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Tue, 25 Oct 2022 15:20:22 -0400 Subject: [PATCH 4/6] lint --- .../java/io/temporal/internal/BackoffThrottler.java | 10 +++++----- .../internal/retryer/GrpcAsyncRetryerTest.java | 8 ++++---- .../temporal/internal/retryer/GrpcSyncRetryerTest.java | 8 ++++---- 3 files changed, 13 insertions(+), 13 deletions(-) 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 59fff8c88e..76647099ac 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java @@ -30,8 +30,7 @@ /** * Used to throttle code execution in presence of failures using exponential backoff logic. * - *

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

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

* @@ -39,9 +38,10 @@ * 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 bahaviour for all SDK. + * 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 bahaviour for all SDK. * *

Example usage: * 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 ec4859fc32..85c199b1ac 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 @@ -80,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 @@ -117,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 @@ -265,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.", @@ -407,6 +407,6 @@ public void testResourceExhaustedFailure() throws InterruptedException { assertTrue("We should retry RESOURCE_EXHAUSTED failures.", attempts.get() > 2); assertTrue( "We should retry RESOURCE_EXHAUSTED failures using longInitialInterval.", - elapsedTime > 2000); + 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 1680fe74a7..93ae2d885c 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 @@ -82,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 @@ -200,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.", @@ -314,8 +314,8 @@ public void testResourceExhaustedFailure() { long elapsedTime = System.currentTimeMillis() - start; assertTrue("We should retry RESOURCE_EXHAUSTED failures.", attempts.get() > 2); assertTrue( - "We should retry RESOURCE_EXHAUSTED failures using longInitialInterval.", - elapsedTime > 2000); + "We should retry RESOURCE_EXHAUSTED failures using congestionInitialInterval.", + elapsedTime >= 2000); } @Test From 428c31f38c9c04b7fe7770fa40a196a76c6d7d54 Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Wed, 26 Oct 2022 17:33:42 -0400 Subject: [PATCH 5/6] Address review comments --- .../io/temporal/internal/worker/Poller.java | 2 +- .../internal/worker/PollerOptions.java | 24 ++++---- .../temporal/internal/BackoffThrottler.java | 24 +++++--- .../internal/retryer/GrpcAsyncRetryer.java | 2 +- .../internal/retryer/GrpcRetryerUtils.java | 2 - .../internal/retryer/GrpcSyncRetryer.java | 2 +- .../serviceclient/RpcRetryOptions.java | 58 ++++++++++--------- .../DefaultStubLongPollRpcRetryOptions.java | 4 +- ...ltStubServiceOperationRpcRetryOptions.java | 4 +- .../retryer/GrpcAsyncRetryerTest.java | 20 +++---- .../internal/retryer/GrpcSyncRetryerTest.java | 18 +++--- 11 files changed, 85 insertions(+), 75 deletions(-) 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 7adf6b7a57..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 @@ -210,7 +210,7 @@ private class PollLoopTask implements Runnable { pollerOptions.getBackoffCongestionInitialInterval(), pollerOptions.getBackoffMaximumInterval(), pollerOptions.getBackoffCoefficient(), - pollerOptions.getBackoffMaximumJitter()); + pollerOptions.getBackoffMaximumJitterCoefficient()); } @Override 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 dd143783d8..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 @@ -58,7 +58,7 @@ public static final class Builder { private Duration backoffInitialInterval = Duration.ofMillis(100); private Duration backoffCongestionInitialInterval = Duration.ofMillis(1000); private Duration backoffMaximumInterval = Duration.ofMinutes(1); - private double backoffMaximumJitter = 0.1; + private double backoffMaximumJitterCoefficient = 0.1; private int pollThreadCount = 1; private String pollThreadNamePrefix; private Thread.UncaughtExceptionHandler uncaughtExceptionHandler; @@ -75,7 +75,7 @@ private Builder(PollerOptions options) { this.backoffInitialInterval = options.getBackoffInitialInterval(); this.backoffCongestionInitialInterval = options.getBackoffCongestionInitialInterval(); this.backoffMaximumInterval = options.getBackoffMaximumInterval(); - this.backoffMaximumJitter = options.getBackoffMaximumJitter(); + this.backoffMaximumJitterCoefficient = options.getBackoffMaximumJitterCoefficient(); this.pollThreadCount = options.getPollThreadCount(); this.pollThreadNamePrefix = options.getPollThreadNamePrefix(); this.uncaughtExceptionHandler = options.getUncaughtExceptionHandler(); @@ -130,8 +130,8 @@ public Builder setBackoffMaximumInterval(Duration backoffMaximumInterval) { * 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 setBackoffMaximumJitter(double backoffMaximumJitter) { - this.backoffMaximumJitter = backoffMaximumJitter; + public Builder setBackoffMaximumJitterCoefficient(double backoffMaximumJitterCoefficient) { + this.backoffMaximumJitterCoefficient = backoffMaximumJitterCoefficient; return this; } @@ -177,7 +177,7 @@ public PollerOptions build() { backoffInitialInterval, backoffCongestionInitialInterval, backoffMaximumInterval, - backoffMaximumJitter, + backoffMaximumJitterCoefficient, pollThreadCount, uncaughtExceptionHandler, pollThreadNamePrefix); @@ -189,7 +189,7 @@ public PollerOptions build() { private final int maximumPollRateIntervalMilliseconds; private final double maximumPollRatePerSecond; private final double backoffCoefficient; - private final double backoffMaximumJitter; + private final double backoffMaximumJitterCoefficient; private final Duration backoffInitialInterval; private final Duration backoffCongestionInitialInterval; private final Duration backoffMaximumInterval; @@ -204,7 +204,7 @@ private PollerOptions( Duration backoffInitialInterval, Duration backoffCongestionInitialInterval, Duration backoffMaximumInterval, - double backoffMaximumJitter, + double backoffMaximumJitterCoefficient, int pollThreadCount, Thread.UncaughtExceptionHandler uncaughtExceptionHandler, String pollThreadNamePrefix) { @@ -214,7 +214,7 @@ private PollerOptions( this.backoffInitialInterval = backoffInitialInterval; this.backoffCongestionInitialInterval = backoffCongestionInitialInterval; this.backoffMaximumInterval = backoffMaximumInterval; - this.backoffMaximumJitter = backoffMaximumJitter; + this.backoffMaximumJitterCoefficient = backoffMaximumJitterCoefficient; this.pollThreadCount = pollThreadCount; this.uncaughtExceptionHandler = uncaughtExceptionHandler; this.pollThreadNamePrefix = pollThreadNamePrefix; @@ -244,8 +244,8 @@ public Duration getBackoffMaximumInterval() { return backoffMaximumInterval; } - public double getBackoffMaximumJitter() { - return backoffMaximumJitter; + public double getBackoffMaximumJitterCoefficient() { + return backoffMaximumJitterCoefficient; } public int getPollThreadCount() { @@ -275,8 +275,8 @@ public String toString() { + backoffCongestionInitialInterval + ", backoffMaximumInterval=" + backoffMaximumInterval - + ", backoffMaximumJitter=" - + backoffMaximumJitter + + ", 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 d650491768..38c9c23143 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java @@ -35,13 +35,14 @@ *

* * 

- * min(pow(backoffCoefficient, failureCount - 1) * initialSleep * (1 + jitter), 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 bahaviour for all SDK. + * occurs after a congestion failure. This is the expected behaviour for all SDK. * *

Example usage: * @@ -79,7 +80,7 @@ public final class BackoffThrottler { private final double backoffCoefficient; - private final double maxJitter; + private final double maxJitterCoefficient; private int failureCount = 0; @@ -92,37 +93,42 @@ public final class BackoffThrottler { * @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 maxJitter maximum jitter to add or subtract + * @param maxJitterCoefficient maximum jitter coefficient (in the range [0.0, 1.0[) to randomly + * add or subtract to sleep time */ public BackoffThrottler( Duration regularInitialSleep, Duration congestionInitialSleep, @Nullable Duration maxSleep, double backoffCoefficient, - double maxJitter) { + 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 (maxJitter < 0 || maxJitter >= 1.0) { - throw new IllegalArgumentException("maximumJitter has to be >= 0 and < 1.0: " + maxJitter); + 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.maxJitter = maxJitter; + this.maxJitterCoefficient = maxJitterCoefficient; } public long getSleepTime() { if (failureCount == 0) return 0; Duration initial = (lastFailureCode == Code.RESOURCE_EXHAUSTED) ? congestionInitialSleep : regularInitialSleep; - double jitter = Math.random() * maxJitter * 2 - maxJitter; + + // 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()); } 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 fbdfb5cdef..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 @@ -70,7 +70,7 @@ public GrpcAsyncRetryer( rpcOptions.getCongestionInitialInterval(), rpcOptions.getMaximumInterval(), rpcOptions.getBackoffCoefficient(), - rpcOptions.getMaximumJitter()); + rpcOptions.getMaximumJitterCoefficient()); } public CompletableFuture retry() { diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryerUtils.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryerUtils.java index 2f9d993723..335bfb97aa 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryerUtils.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryerUtils.java @@ -76,8 +76,6 @@ class GrpcRetryerUtils { // By default, we keep retrying with DEADLINE_EXCEEDED assuming that it's the deadline of // one attempt which expired, but not the whole sequence. break; - case RESOURCE_EXHAUSTED: - break; default: for (RpcRetryOptions.DoNotRetryItem pair : options.getDoNotRetry()) { if (pair.getCode() == code 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 e4ff29cf33..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 @@ -51,7 +51,7 @@ public R retry( rpcOptions.getCongestionInitialInterval(), rpcOptions.getMaximumInterval(), rpcOptions.getBackoffCoefficient(), - rpcOptions.getMaximumJitter()); + rpcOptions.getMaximumJitterCoefficient()); int attempt = 0; StatusRuntimeException lastMeaningfulException = null; 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 e88ee64bdb..60c1ae02d3 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java @@ -95,9 +95,9 @@ public static final class Builder { private Duration maximumInterval; - // 0 is a valid value for maximumJitter, and yet, it is not the default value for maximumJitter. - // Use NaN instead as a marker for not configured jitter - private double maximumJitter = Double.NaN; + // 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<>(); @@ -113,7 +113,7 @@ private Builder(RpcRetryOptions options) { this.initialInterval = options.getInitialInterval(); this.congestionInitialInterval = options.getCongestionInitialInterval(); this.maximumInterval = options.getMaximumInterval(); - this.maximumJitter = options.getMaximumJitter(); + this.maximumJitterCoefficient = options.getMaximumJitterCoefficient(); this.doNotRetry = options.getDoNotRetry(); } @@ -210,12 +210,15 @@ public Builder setMaximumInterval(Duration maximumInterval) { * 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 setMaximumJitter(double maximumJitter) { - if (!Double.isFinite(maximumJitter) || maximumJitter < 0 || maximumJitter >= 1.0) { + public Builder setMaximumJitterCoefficient(double maximumJitterCoefficient) { + if ((!Double.isFinite(maximumJitterCoefficient) + || maximumJitterCoefficient < 0 + || maximumJitterCoefficient >= 1.0) + && maximumJitterCoefficient != -1.0) { throw new IllegalArgumentException( - "maximumJitter has to be >= 0 and < 1.0: " + maximumJitter); + "maximumJitterCoefficient has to be >= 0 and < 1.0: " + maximumJitterCoefficient); } - this.maximumJitter = maximumJitter; + this.maximumJitterCoefficient = maximumJitterCoefficient; return this; } @@ -287,8 +290,8 @@ public Builder setRetryOptions(RpcRetryOptions o) { setBackoffCoefficient( OptionsUtils.merge(backoffCoefficient, o.getBackoffCoefficient(), double.class)); setMaximumAttempts(OptionsUtils.merge(maximumAttempts, o.getMaximumAttempts(), int.class)); - if (Double.isFinite(o.getMaximumJitter())) { - setMaximumJitter(o.getMaximumJitter()); + if (o.getMaximumJitterCoefficient() != -1.0) { + setMaximumJitterCoefficient(o.getMaximumJitterCoefficient()); } doNotRetry = merge(doNotRetry, o.getDoNotRetry()); validateBuildWithDefaults(); @@ -317,7 +320,7 @@ public RpcRetryOptions build() { expiration, maximumAttempts, maximumInterval, - maximumJitter, + maximumJitterCoefficient, doNotRetry); } @@ -356,8 +359,9 @@ public RpcRetryOptions validateBuildWithDefaults() { } } - if (Double.isNaN(maximumJitter)) { - maximumJitter = DefaultStubServiceOperationRpcRetryOptions.MAXIMUM_JITTER; + if (maximumJitterCoefficient == -1.0) { + maximumJitterCoefficient = + DefaultStubServiceOperationRpcRetryOptions.MAXIMUM_JITTER_COEFFICIENT; } RpcRetryOptions result = @@ -368,7 +372,7 @@ public RpcRetryOptions validateBuildWithDefaults() { expiration, maximumAttempts, maxInterval, - maximumJitter, + maximumJitterCoefficient, MoreObjects.firstNonNull(doNotRetry, Collections.emptyList())); result.validate(); return result; @@ -387,7 +391,7 @@ public RpcRetryOptions validateBuildWithDefaults() { private final Duration maximumInterval; - private final double maximumJitter; + private final double maximumJitterCoefficient; private final @Nonnull List doNotRetry; @@ -398,7 +402,7 @@ private RpcRetryOptions( Duration expiration, int maximumAttempts, Duration maximumInterval, - double maximumJitter, + double maximumJitterCoefficient, @Nonnull List doNotRetry) { this.initialInterval = initialInterval; this.congestionInitialInterval = congestionInitialInterval; @@ -406,7 +410,7 @@ private RpcRetryOptions( this.expiration = expiration; this.maximumAttempts = maximumAttempts; this.maximumInterval = maximumInterval; - this.maximumJitter = maximumJitter; + this.maximumJitterCoefficient = maximumJitterCoefficient; this.doNotRetry = Collections.unmodifiableList(doNotRetry); } @@ -434,8 +438,8 @@ public Duration getMaximumInterval() { return maximumInterval; } - public double getMaximumJitter() { - return maximumJitter; + public double getMaximumJitterCoefficient() { + return maximumJitterCoefficient; } public void validate() { @@ -447,7 +451,7 @@ public void validate(boolean hasToBeFinite) { throw new IllegalStateException("required property initialInterval not set"); } if (congestionInitialInterval == null) { - throw new IllegalStateException("required property longInitialInterval not set"); + throw new IllegalStateException("required property congestionInitialInterval not set"); } if (maximumAttempts < 0) { throw new IllegalArgumentException("negative maximum attempts"); @@ -466,9 +470,11 @@ public void validate(boolean hasToBeFinite) { if (backoffCoefficient != 0d && backoffCoefficient < 1.0) { throw new IllegalArgumentException("coefficient less than 1"); } - if (!Double.isFinite(maximumJitter) || maximumJitter < 0 || maximumJitter >= 1.0) { + if (!Double.isFinite(maximumJitterCoefficient) + || maximumJitterCoefficient < 0 + || maximumJitterCoefficient >= 1.0) { throw new IllegalArgumentException( - "maximumJitter has to be >= 0 and < 1.0: " + maximumJitter); + "maximumJitterCoefficient has to be >= 0 and < 1.0: " + maximumJitterCoefficient); } } @@ -491,8 +497,8 @@ public String toString() { + maximumAttempts + ", maximumInterval=" + maximumInterval - + ", maximumJitter=" - + maximumJitter + + ", maximumJitterCoefficient=" + + maximumJitterCoefficient + ", doNotRetry=" + doNotRetry + '}'; @@ -509,7 +515,7 @@ public boolean equals(Object o) { && Objects.equals(congestionInitialInterval, that.congestionInitialInterval) && Objects.equals(expiration, that.expiration) && Objects.equals(maximumInterval, that.maximumInterval) - && Objects.equals(maximumJitter, that.maximumJitter) + && Objects.equals(maximumJitterCoefficient, that.maximumJitterCoefficient) && Objects.equals(doNotRetry, that.doNotRetry); } @@ -522,7 +528,7 @@ public int hashCode() { expiration, maximumAttempts, maximumInterval, - maximumJitter, + 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 ba76db72a3..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 @@ -29,7 +29,7 @@ public class DefaultStubLongPollRpcRetryOptions { 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 = 0.1; + 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(); @@ -47,7 +47,7 @@ private static RpcRetryOptions.Builder getBuilder() { .setCongestionInitialInterval(CONGESTION_INITIAL_INTERVAL) .setBackoffCoefficient(BACKOFF) .setMaximumInterval(MAXIMUM_INTERVAL) - .setMaximumJitter(MAXIMUM_JITTER); + .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 399ad0803f..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 @@ -34,7 +34,7 @@ public class DefaultStubServiceOperationRpcRetryOptions { 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 = 0.1; + public static final double MAXIMUM_JITTER_COEFFICIENT = 0.1; public static final RpcRetryOptions INSTANCE; @@ -55,6 +55,6 @@ public static RpcRetryOptions.Builder getBuilder() { .setExpiration(EXPIRATION_INTERVAL) .setBackoffCoefficient(BACKOFF) .setMaximumInterval(MAXIMUM_INTERVAL) - .setMaximumJitter(MAXIMUM_JITTER); + .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 85c199b1ac..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,7 +58,7 @@ public void testExpirationAsync() throws InterruptedException { .setInitialInterval(Duration.ofMillis(10)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -92,7 +92,7 @@ public void testExpirationFutureAsync() throws InterruptedException { .setInitialInterval(Duration.ofMillis(10)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -128,7 +128,7 @@ public void testDoNotRetryAsync() throws InterruptedException { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .addDoNotRetry(STATUS_CODE, null) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); @@ -164,7 +164,7 @@ public void testInterruptedExceptionAsync() throws InterruptedException { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -197,7 +197,7 @@ public void testNotStatusRuntimeExceptionAsync() throws InterruptedException { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -232,7 +232,7 @@ public void testRetryDeadlineExceededException() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -279,7 +279,7 @@ public void testRespectGlobalDeadlineExceeded() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(50000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -328,7 +328,7 @@ public void testGlobalDeadlineExceededAfterAnotherException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); final AtomicReference exception = new AtomicReference<>(); @@ -375,7 +375,7 @@ public void testResourceExhaustedFailure() throws InterruptedException { .setInitialInterval(Duration.ofMillis(1)) .setCongestionInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .setMaximumAttempts(3) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); @@ -406,7 +406,7 @@ public void testResourceExhaustedFailure() throws InterruptedException { long elapsedTime = System.currentTimeMillis() - start; assertTrue("We should retry RESOURCE_EXHAUSTED failures.", attempts.get() > 2); assertTrue( - "We should retry RESOURCE_EXHAUSTED failures using longInitialInterval.", + "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 93ae2d885c..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,7 +63,7 @@ public void testExpiration() { .setInitialInterval(Duration.ofMillis(10)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -93,7 +93,7 @@ public void testDoNotRetry() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .addDoNotRetry(STATUS_CODE, null) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); @@ -123,7 +123,7 @@ public void testInterruptedException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -149,7 +149,7 @@ public void testNotStatusRuntimeException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -179,7 +179,7 @@ public void testRetryDeadlineExceededException() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(500)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -214,7 +214,7 @@ public void testRespectGlobalDeadlineExceeded() { .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) .setExpiration(Duration.ofMillis(50000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); final AtomicInteger attempts = new AtomicInteger(); @@ -252,7 +252,7 @@ public void testGlobalDeadlineExceededAfterAnotherException() { RpcRetryOptions.newBuilder() .setInitialInterval(Duration.ofMillis(100)) .setMaximumInterval(Duration.ofMillis(100)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .validateBuildWithDefaults(); final AtomicReference exception = new AtomicReference<>(); @@ -291,7 +291,7 @@ public void testResourceExhaustedFailure() { .setInitialInterval(Duration.ofMillis(1)) .setCongestionInitialInterval(Duration.ofMillis(1000)) .setMaximumInterval(Duration.ofMillis(1000)) - .setMaximumJitter(0) + .setMaximumJitterCoefficient(0) .setMaximumAttempts(3) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); @@ -336,6 +336,6 @@ public void testCongestionAndJitterAreNotMandatory() { // The following were added latter; they must silently use default values if unspecified assertEquals(Duration.ofMillis(1000), options.getCongestionInitialInterval()); - assertEquals(0.1, options.getMaximumJitter(), 0.01); + assertEquals(0.1, options.getMaximumJitterCoefficient(), 0.01); } } From d9d69d2ac760d89c01c29e1394bd2849fb6c91ee Mon Sep 17 00:00:00 2001 From: James Watkins-Harvey Date: Thu, 27 Oct 2022 13:11:54 -0400 Subject: [PATCH 6/6] Address more review comments --- .../serviceclient/RpcRetryOptions.java | 88 +++++++++++++------ 1 file changed, 60 insertions(+), 28 deletions(-) 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 60c1ae02d3..f09be6b823 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/RpcRetryOptions.java @@ -119,35 +119,50 @@ private Builder(RpcRetryOptions options) { /** * Interval of the first retry, on regular failures. If coefficient is 1.0 then it is used for - * all retries. Required. + * 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; } - this.initialInterval = initialInterval; 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) { - Objects.requireNonNull(congestionInitialInterval); - if (congestionInitialInterval.isNegative() || congestionInitialInterval.isZero()) { - throw new IllegalArgumentException("Invalid interval: " + congestionInitialInterval); + if (initialInterval != null) { + if (congestionInitialInterval.isNegative() || congestionInitialInterval.isZero()) { + throw new IllegalArgumentException("Invalid interval: " + congestionInitialInterval); + } + this.congestionInitialInterval = congestionInitialInterval; + } else { + this.congestionInitialInterval = null; } - this.congestionInitialInterval = congestionInitialInterval; 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) { @@ -164,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; } @@ -177,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) { @@ -195,30 +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; } - this.maximumInterval = maximumInterval; 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 ((!Double.isFinite(maximumJitterCoefficient) - || maximumJitterCoefficient < 0 - || maximumJitterCoefficient >= 1.0) - && maximumJitterCoefficient != -1.0) { - throw new IllegalArgumentException( - "maximumJitterCoefficient has to be >= 0 and < 1.0: " + 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.maximumJitterCoefficient = maximumJitterCoefficient; return this; }