Use longer retry interval on RESOURCE_EXHAUSTED

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 {

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,27 +188,33 @@ 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;
 
   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='"

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.
+ *
+ * <p>The formula used to calculate the next sleep interval is:
  *
  * <p>
  *
  * <pre>
- * 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);
  * </pre>
  *
+ * where <code>initialSleep</code> is either set to <code>regularInitialSleep</code> or <code>
+ * congestionInitialSleep</code> based on the <i>most recent</i> 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.
+ *
  * <p>Example usage:
  *
  * <pre><code>
- * 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);
  *     }
  * }
  * </code></pre>
@@ -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;
   }
 }