From 1399c63ae5481a8cb05f5e9f6a907fc7a098e367 Mon Sep 17 00:00:00 2001
From: James Watkins-Harvey <mjameswh@gmail.com>
Date: Fri, 30 Sep 2022 13:45:01 -0400
Subject: [PATCH 1/4] Use BackoffThrottler only from a single thread

---
 .../io/temporal/internal/worker/Poller.java   |  16 ++-
 .../internal/AsyncBackoffThrottler.java       | 130 ------------------
 .../temporal/internal/BackoffThrottler.java   |  52 ++++++-
 .../internal/retryer/GrpcAsyncRetryer.java    |  12 +-
 4 files changed, 60 insertions(+), 150 deletions(-)
 delete mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/AsyncBackoffThrottler.java

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 c27a5bb074..a396fa201d 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
@@ -62,7 +62,6 @@ interface ThrowingRunnable {
 
   private final AtomicReference<CountDownLatch> suspendLatch = new AtomicReference<>();
 
-  private BackoffThrottler pollBackoffThrottler;
   private Throttler pollRateThrottler;
 
   private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler =
@@ -113,11 +112,6 @@ public void start() {
         new ExecutorThreadFactory(
             pollerOptions.getPollThreadNamePrefix(), pollerOptions.getUncaughtExceptionHandler()));
 
-    pollBackoffThrottler =
-        new BackoffThrottler(
-            pollerOptions.getPollBackoffInitialInterval(),
-            pollerOptions.getPollBackoffMaximumInterval(),
-            pollerOptions.getPollBackoffCoefficient());
     for (int i = 0; i < pollerOptions.getPollThreadCount(); i++) {
       pollExecutor.execute(new PollLoopTask(new PollExecutionTask()));
       workerMetricsScope.counter(MetricsType.POLLER_START_COUNTER).inc(1);
@@ -206,9 +200,15 @@ public String toString() {
   private class PollLoopTask implements Runnable {
 
     private final Poller.ThrowingRunnable task;
+    private final BackoffThrottler pollBackoffThrottler;
 
     PollLoopTask(Poller.ThrowingRunnable task) {
       this.task = task;
+      this.pollBackoffThrottler =
+          new BackoffThrottler(
+              pollerOptions.getPollBackoffInitialInterval(),
+              pollerOptions.getPollBackoffMaximumInterval(),
+              pollerOptions.getPollBackoffCoefficient());
     }
 
     @Override
@@ -237,8 +237,10 @@ public void run() {
         if (e instanceof InterruptedException) {
           // we restore the flag here, so it can be checked and processed (with exit) in finally.
           Thread.currentThread().interrupt();
+        } else {
+          // Don't increase throttle on InterruptedException
+          pollBackoffThrottler.failure();
         }
-        pollBackoffThrottler.failure();
         uncaughtExceptionHandler.uncaughtException(Thread.currentThread(), e);
       } finally {
         if (!shouldTerminate()) {
diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/AsyncBackoffThrottler.java b/temporal-serviceclient/src/main/java/io/temporal/internal/AsyncBackoffThrottler.java
deleted file mode 100644
index dffff83383..0000000000
--- a/temporal-serviceclient/src/main/java/io/temporal/internal/AsyncBackoffThrottler.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/*
- * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
- *
- * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- *
- * Modifications copyright (C) 2017 Uber Technologies, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this material except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.temporal.internal;
-
-import io.grpc.Context;
-import java.time.Duration;
-import java.util.Objects;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * Used to throttle code execution in presence of failures using exponential backoff logic. The
- * formula used to calculate the next sleep interval is:
- *
- * <p>
- *
- * <pre>
- * min(pow(backoffCoefficient, failureCount - 1) * initialSleep, maxSleep);
- * </pre>
- *
- * <p>Example usage:
- *
- * <p>
- *
- * <pre>
- * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
- * while(!stopped) {
- *     try {
- *         Future&lt;Void&gt; t = throttler.throttle();
- *         t.get();
- *         // some code that can fail and should be throttled
- *         ...
- *         throttler.success();
- *     }
- *     catch (Exception e) {
- *         throttler.failure();
- *     }
- * }
- * </pre>
- *
- * @author fateev
- */
-public final class AsyncBackoffThrottler {
-
-  private static final ScheduledExecutorService executor =
-      new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "async-backoff-throttler"));
-
-  private final Duration initialSleep;
-
-  private final Duration maxSleep;
-
-  private final double backoffCoefficient;
-
-  private final AtomicLong failureCount = new AtomicLong();
-
-  /**
-   * Construct an instance of the throttler.
-   *
-   * @param initialSleep time to sleep on the first failure
-   * @param maxSleep maximum time to sleep independently of number of failures
-   * @param backoffCoefficient coefficient used to calculate the next time to sleep.
-   */
-  public AsyncBackoffThrottler(
-      Duration initialSleep, Duration maxSleep, double backoffCoefficient) {
-    Objects.requireNonNull(initialSleep, "initialSleep");
-    if (backoffCoefficient < 1.0) {
-      throw new IllegalArgumentException(
-          "backoff coefficient less than 1.0: " + backoffCoefficient);
-    }
-    this.initialSleep = initialSleep;
-    this.maxSleep = maxSleep;
-    this.backoffCoefficient = backoffCoefficient;
-  }
-
-  private long calculateSleepTime() {
-    double sleepMillis =
-        Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis();
-    if (maxSleep != null) {
-      return Math.min((long) sleepMillis, maxSleep.toMillis());
-    }
-    return (long) sleepMillis;
-  }
-
-  /** Result future is done after a delay if there were failures since the last success call. */
-  public CompletableFuture<Void> throttle() {
-    if (failureCount.get() == 0) {
-      return CompletableFuture.completedFuture(null);
-    }
-    CompletableFuture<Void> result = new CompletableFuture<>();
-    long delay = calculateSleepTime();
-    @SuppressWarnings({"FutureReturnValueIgnored", "unused"})
-    ScheduledFuture<?> ignored =
-        executor.schedule(
-            // preserving gRPC context between threads
-            Context.current().wrap(() -> result.complete(null)), delay, TimeUnit.MILLISECONDS);
-    return result;
-  }
-
-  /** Resent failure count to 0. */
-  public void success() {
-    failureCount.set(0);
-  }
-
-  /** Increment failure count. */
-  public void failure() {
-    failureCount.incrementAndGet();
-  }
-}
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 e7503d769b..99f263623f 100644
--- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
+++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
@@ -20,9 +20,11 @@
 
 package io.temporal.internal;
 
+import io.grpc.Context;
+
 import java.time.Duration;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.*;
 import javax.annotation.Nullable;
 
 /**
@@ -54,17 +56,38 @@
  * }
  * </pre>
  *
+ * <p>or using async semantic
+ *
+ * <pre>
+ * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
+ * while(!stopped) {
+ *     try {
+ *         Future&lt;Void&gt; t = throttler.throttleAsync();
+ *         t.get();
+ *         // some code that can fail and should be throttled
+ *         ...
+ *         throttler.success();
+ *     }
+ *     catch (Exception e) {
+ *         throttler.failure();
+ *     }
+ * }
+ * </pre>
+ *
  * @author fateev
  */
 public final class BackoffThrottler {
 
+  private static final ScheduledExecutorService executor =
+      new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "async-backoff-throttler"));
+
   private final Duration initialSleep;
 
   private final Duration maxSleep;
 
   private final double backoffCoefficient;
 
-  private final AtomicLong failureCount = new AtomicLong();
+  private int failureCount = 0;
 
   /**
    * Construct an instance of the throttler.
@@ -83,7 +106,7 @@ public BackoffThrottler(
 
   private long calculateSleepTime() {
     double sleepMillis =
-        Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis();
+        Math.pow(backoffCoefficient, failureCount - 1) * initialSleep.toMillis();
     if (maxSleep != null) {
       return Math.min((long) sleepMillis, maxSleep.toMillis());
     }
@@ -96,18 +119,33 @@ private long calculateSleepTime() {
    * @throws InterruptedException
    */
   public void throttle() throws InterruptedException {
-    if (failureCount.get() > 0) {
+    if (failureCount > 0) {
       Thread.sleep(calculateSleepTime());
     }
   }
 
-  /** Resent failure count to 0. */
+  /** Result future is done after a delay if there were failures since the last success call. */
+  public CompletableFuture<Void> throttleAsync() {
+    if (failureCount == 0) {
+      return CompletableFuture.completedFuture(null);
+    }
+    CompletableFuture<Void> result = new CompletableFuture<>();
+    long delay = calculateSleepTime();
+    @SuppressWarnings({"FutureReturnValueIgnored", "unused"})
+    ScheduledFuture<?> ignored =
+        executor.schedule(
+            // preserving gRPC context between threads
+            Context.current().wrap(() -> result.complete(null)), delay, TimeUnit.MILLISECONDS);
+    return result;
+  }
+
+  /** Reset failure count to 0. */
   public void success() {
-    failureCount.set(0);
+    failureCount = 0;
   }
 
   /** Increment failure count. */
   public void failure() {
-    failureCount.incrementAndGet();
+    failureCount++;
   }
 }
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 ec64df2c2a..62188434cb 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
@@ -24,7 +24,7 @@
 import io.grpc.Deadline;
 import io.grpc.StatusRuntimeException;
 import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
-import io.temporal.internal.AsyncBackoffThrottler;
+import io.temporal.internal.BackoffThrottler;
 import io.temporal.serviceclient.RpcRetryOptions;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -46,8 +46,8 @@ public <R> CompletableFuture<R> retry(
     @Nullable
     Deadline retriesExpirationDeadline =
         GrpcRetryerUtils.mergeDurationWithAnAbsoluteDeadline(rpcOptions.getExpiration(), deadline);
-    AsyncBackoffThrottler throttler =
-        new AsyncBackoffThrottler(
+    BackoffThrottler throttler =
+        new BackoffThrottler(
             rpcOptions.getInitialInterval(),
             rpcOptions.getMaximumInterval(),
             rpcOptions.getBackoffCoefficient());
@@ -72,11 +72,11 @@ private <R> void retry(
       Supplier<CompletableFuture<R>> function,
       int attempt,
       @Nullable Deadline retriesExpirationDeadline,
-      AsyncBackoffThrottler throttler,
+      BackoffThrottler throttler,
       StatusRuntimeException previousException,
       CompletableFuture<R> resultCF) {
     throttler
-        .throttle()
+        .throttleAsync()
         .thenAccept(
             (ignore) -> {
               if (previousException != null) {
@@ -140,7 +140,7 @@ private <R> void failOrRetry(
       Supplier<CompletableFuture<R>> function,
       int attempt,
       @Nullable Deadline retriesExpirationDeadline,
-      AsyncBackoffThrottler throttler,
+      BackoffThrottler throttler,
       StatusRuntimeException previousException,
       Throwable currentException,
       CompletableFuture<R> resultCF) {

From c1052a39a9ca186812a683a81d2852b2bc611e03 Mon Sep 17 00:00:00 2001
From: James Watkins-Harvey <mjameswh@gmail.com>
Date: Fri, 30 Sep 2022 14:49:20 -0400
Subject: [PATCH 2/4] lint

---
 .../src/main/java/io/temporal/internal/BackoffThrottler.java  | 4 +---
 1 file changed, 1 insertion(+), 3 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 99f263623f..ebf75b721d 100644
--- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
+++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
@@ -21,7 +21,6 @@
 package io.temporal.internal;
 
 import io.grpc.Context;
-
 import java.time.Duration;
 import java.util.Objects;
 import java.util.concurrent.*;
@@ -105,8 +104,7 @@ public BackoffThrottler(
   }
 
   private long calculateSleepTime() {
-    double sleepMillis =
-        Math.pow(backoffCoefficient, failureCount - 1) * initialSleep.toMillis();
+    double sleepMillis = Math.pow(backoffCoefficient, failureCount - 1) * initialSleep.toMillis();
     if (maxSleep != null) {
       return Math.min((long) sleepMillis, maxSleep.toMillis());
     }

From 182ba40af5eb50080f5816f741fe71e7e128d88c Mon Sep 17 00:00:00 2001
From: James Watkins-Harvey <mjameswh@gmail.com>
Date: Wed, 12 Oct 2022 16:27:35 -0400
Subject: [PATCH 3/4] Address review comments

---
 .../external/GenericWorkflowClientImpl.java   |   4 +
 .../io/temporal/internal/worker/Poller.java   |   5 +-
 .../temporal/internal/BackoffThrottler.java   |  57 +-----
 .../internal/retryer/GrpcAsyncRetryer.java    | 183 ++++++++----------
 .../internal/retryer/GrpcRetryer.java         |  11 +-
 .../internal/retryer/GrpcSyncRetryer.java     |   8 +-
 .../retryer/GrpcAsyncRetryerTest.java         |  78 +++++---
 .../internal/retryer/GrpcSyncRetryerTest.java |  35 ++--
 8 files changed, 176 insertions(+), 205 deletions(-)

diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java
index 46e4adca1c..f5e41f1e6a 100644
--- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java
+++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java
@@ -40,6 +40,9 @@
 
 public final class GenericWorkflowClientImpl implements GenericWorkflowClient {
 
+  private static final ScheduledExecutorService executor =
+      new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "generic-wf-client-async-throttler"));
+
   private final WorkflowServiceStubs service;
   private final Scope metricsScope;
   private final GrpcRetryer grpcRetryer;
@@ -160,6 +163,7 @@ public GetWorkflowExecutionHistoryResponse longPollHistory(
   public CompletableFuture<GetWorkflowExecutionHistoryResponse> longPollHistoryAsync(
       @Nonnull GetWorkflowExecutionHistoryRequest request, @Nonnull Deadline deadline) {
     return grpcRetryer.retryWithResultAsync(
+        executor,
         () -> {
           CompletableFuture<GetWorkflowExecutionHistoryResponse> result = new CompletableFuture<>();
           ListenableFuture<GetWorkflowExecutionHistoryResponse> resultFuture =
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 a396fa201d..6bbd3b6601 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
@@ -214,7 +214,10 @@ private class PollLoopTask implements Runnable {
     @Override
     public void run() {
       try {
-        pollBackoffThrottler.throttle();
+        long throttleMs = pollBackoffThrottler.getSleepTime();
+        if (throttleMs > 0) {
+          Thread.sleep(throttleMs);
+        }
         if (pollRateThrottler != null) {
           pollRateThrottler.throttle();
         }
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 ebf75b721d..5e741cd09d 100644
--- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
+++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
@@ -20,10 +20,8 @@
 
 package io.temporal.internal;
 
-import io.grpc.Context;
 import java.time.Duration;
 import java.util.Objects;
-import java.util.concurrent.*;
 import javax.annotation.Nullable;
 
 /**
@@ -44,25 +42,10 @@
  * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
  * while(!stopped) {
  *     try {
- *         throttler.throttle();
- *         // some code that can fail and should be throttled
- *         ...
- *         throttler.success();
- *     }
- *     catch (Exception e) {
- *         throttler.failure();
- *     }
- * }
- * </pre>
- *
- * <p>or using async semantic
- *
- * <pre>
- * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
- * while(!stopped) {
- *     try {
- *         Future&lt;Void&gt; t = throttler.throttleAsync();
- *         t.get();
+ *         long throttleMs = throttler.getSleepTime();
+ *         if (throttleMs > 0) {
+ *             Thread.sleep(throttleMs);
+ *         }
  *         // some code that can fail and should be throttled
  *         ...
  *         throttler.success();
@@ -77,9 +60,6 @@
  */
 public final class BackoffThrottler {
 
-  private static final ScheduledExecutorService executor =
-      new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "async-backoff-throttler"));
-
   private final Duration initialSleep;
 
   private final Duration maxSleep;
@@ -103,7 +83,8 @@ public BackoffThrottler(
     this.backoffCoefficient = backoffCoefficient;
   }
 
-  private long calculateSleepTime() {
+  public long getSleepTime() {
+    if (failureCount == 0) return 0;
     double sleepMillis = Math.pow(backoffCoefficient, failureCount - 1) * initialSleep.toMillis();
     if (maxSleep != null) {
       return Math.min((long) sleepMillis, maxSleep.toMillis());
@@ -111,30 +92,8 @@ private long calculateSleepTime() {
     return (long) sleepMillis;
   }
 
-  /**
-   * Sleep if there were failures since the last success call.
-   *
-   * @throws InterruptedException
-   */
-  public void throttle() throws InterruptedException {
-    if (failureCount > 0) {
-      Thread.sleep(calculateSleepTime());
-    }
-  }
-
-  /** Result future is done after a delay if there were failures since the last success call. */
-  public CompletableFuture<Void> throttleAsync() {
-    if (failureCount == 0) {
-      return CompletableFuture.completedFuture(null);
-    }
-    CompletableFuture<Void> result = new CompletableFuture<>();
-    long delay = calculateSleepTime();
-    @SuppressWarnings({"FutureReturnValueIgnored", "unused"})
-    ScheduledFuture<?> ignored =
-        executor.schedule(
-            // preserving gRPC context between threads
-            Context.current().wrap(() -> result.complete(null)), delay, TimeUnit.MILLISECONDS);
-    return result;
+  public int getAttemptCount() {
+    return failureCount;
   }
 
   /** Reset failure count to 0. */
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 62188434cb..ee5ebbe54d 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
@@ -28,122 +28,99 @@
 import io.temporal.serviceclient.RpcRetryOptions;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
-import javax.annotation.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-class GrpcAsyncRetryer {
+class GrpcAsyncRetryer<R> {
   private static final Logger log = LoggerFactory.getLogger(GrpcRetryer.class);
 
-  public <R> CompletableFuture<R> retry(
+  private final ScheduledExecutorService executor;
+  private final GrpcRetryer.GrpcRetryerOptions options;
+  private final GetSystemInfoResponse.Capabilities serverCapabilities;
+  private final Supplier<CompletableFuture<R>> function;
+  private final BackoffThrottler throttler;
+  private final Deadline retriesExpirationDeadline;
+  private StatusRuntimeException lastMeaningfulException = null;
+
+  public GrpcAsyncRetryer(
+      ScheduledExecutorService executor,
       Supplier<CompletableFuture<R>> function,
       GrpcRetryer.GrpcRetryerOptions options,
       GetSystemInfoResponse.Capabilities serverCapabilities) {
+
     options.validate();
+
+    this.executor = executor;
+    this.options = options;
+    this.serverCapabilities = serverCapabilities;
+    this.function = function;
+
     RpcRetryOptions rpcOptions = options.getOptions();
-    @Nullable Deadline deadline = options.getDeadline();
-    @Nullable
-    Deadline retriesExpirationDeadline =
-        GrpcRetryerUtils.mergeDurationWithAnAbsoluteDeadline(rpcOptions.getExpiration(), deadline);
-    BackoffThrottler throttler =
+    this.retriesExpirationDeadline =
+        GrpcRetryerUtils.mergeDurationWithAnAbsoluteDeadline(
+            rpcOptions.getExpiration(), options.getDeadline());
+    this.throttler =
         new BackoffThrottler(
             rpcOptions.getInitialInterval(),
             rpcOptions.getMaximumInterval(),
             rpcOptions.getBackoffCoefficient());
+  }
 
-    int attempt = 1;
+  public CompletableFuture<R> retry() {
     CompletableFuture<R> resultCF = new CompletableFuture<>();
-    retry(
-        options,
-        serverCapabilities,
-        function,
-        attempt,
-        retriesExpirationDeadline,
-        throttler,
-        null,
-        resultCF);
+    retry(resultCF);
     return resultCF;
   }
 
-  private <R> void retry(
-      GrpcRetryer.GrpcRetryerOptions options,
-      GetSystemInfoResponse.Capabilities serverCapabilities,
-      Supplier<CompletableFuture<R>> function,
-      int attempt,
-      @Nullable Deadline retriesExpirationDeadline,
-      BackoffThrottler throttler,
-      StatusRuntimeException previousException,
-      CompletableFuture<R> resultCF) {
-    throttler
-        .throttleAsync()
-        .thenAccept(
-            (ignore) -> {
-              if (previousException != null) {
-                log.debug("Retrying after failure", previousException);
-              }
-
-              // try-catch is because get() call might throw.
-              CompletableFuture<R> result;
-
-              try {
-                result = function.get();
-              } catch (Throwable e) {
-                throttler.failure();
-                // 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
-                // completable future
-                // Do not retry if it's not StatusRuntimeException
-                failOrRetry(
-                    options,
-                    serverCapabilities,
-                    function,
-                    attempt,
-                    retriesExpirationDeadline,
-                    throttler,
-                    previousException,
-                    e,
-                    resultCF);
-                return;
-              }
-              if (result == null) {
-                resultCF.complete(null);
-                return;
-              }
-
-              result.whenComplete(
-                  (r, e) -> {
-                    if (e == null) {
-                      throttler.success();
-                      resultCF.complete(r);
-                    } else {
-                      throttler.failure();
-                      failOrRetry(
-                          options,
-                          serverCapabilities,
-                          function,
-                          attempt,
-                          retriesExpirationDeadline,
-                          throttler,
-                          previousException,
-                          e,
-                          resultCF);
-                    }
-                  });
-            });
+  private void retry(CompletableFuture<R> resultCF) {
+    CompletableFuture<Void> throttleFuture = new CompletableFuture<>();
+    @SuppressWarnings({"FutureReturnValueIgnored", "unused"})
+    ScheduledFuture<?> ignored =
+        executor.schedule(
+            // preserving gRPC context between threads
+            Context.current().wrap(() -> throttleFuture.complete(null)),
+            throttler.getSleepTime(),
+            TimeUnit.MILLISECONDS);
+
+    throttleFuture.thenAccept(
+        (ignore) -> {
+          if (lastMeaningfulException != null) {
+            log.debug("Retrying after failure", lastMeaningfulException);
+          }
+
+          // try-catch is because get() call might throw.
+          try {
+            CompletableFuture<R> result = function.get();
+            if (result == null) result = CompletableFuture.completedFuture(null);
+
+            result.whenComplete(
+                (r, e) -> {
+                  if (e == null) {
+                    throttler.success();
+                    resultCF.complete(r);
+                  } else {
+                    throttler.failure();
+                    failOrRetry(e, resultCF);
+                  }
+                });
+
+          } catch (Throwable e) {
+            throttler.failure();
+            // 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
+            // completable future
+            // Do not retry if it's not StatusRuntimeException
+            failOrRetry(e, resultCF);
+          }
+        });
   }
 
-  private <R> void failOrRetry(
-      GrpcRetryer.GrpcRetryerOptions options,
-      GetSystemInfoResponse.Capabilities serverCapabilities,
-      Supplier<CompletableFuture<R>> function,
-      int attempt,
-      @Nullable Deadline retriesExpirationDeadline,
-      BackoffThrottler throttler,
-      StatusRuntimeException previousException,
-      Throwable currentException,
-      CompletableFuture<R> resultCF) {
+  private void failOrRetry(Throwable currentException, CompletableFuture<R> resultCF) {
 
     // If exception is thrown from CompletionStage/CompletableFuture methods like compose or handle
     // - it gets wrapped into CompletionException, so here we need to unwrap it. We can get not
@@ -168,25 +145,17 @@ private <R> void failOrRetry(
       return;
     }
 
-    StatusRuntimeException lastMeaningfulException =
-        GrpcRetryerUtils.lastMeaningfulException(statusRuntimeException, previousException);
+    this.lastMeaningfulException =
+        GrpcRetryerUtils.lastMeaningfulException(statusRuntimeException, lastMeaningfulException);
     if (GrpcRetryerUtils.ranOutOfRetries(
         options.getOptions(),
-        attempt,
-        retriesExpirationDeadline,
+        this.throttler.getAttemptCount(),
+        this.retriesExpirationDeadline,
         Context.current().getDeadline())) {
       log.debug("Out of retries, throwing", lastMeaningfulException);
       resultCF.completeExceptionally(lastMeaningfulException);
     } else {
-      retry(
-          options,
-          serverCapabilities,
-          function,
-          attempt + 1,
-          retriesExpirationDeadline,
-          throttler,
-          lastMeaningfulException,
-          resultCF);
+      retry(resultCF);
     }
   }
 
diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java
index 346d56ce81..4fbfb567ab 100644
--- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java
+++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java
@@ -25,13 +25,12 @@
 import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
 import io.temporal.serviceclient.RpcRetryOptions;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Supplier;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 public final class GrpcRetryer {
-  private static final GrpcSyncRetryer SYNC = new GrpcSyncRetryer();
-  private static final GrpcAsyncRetryer ASYNC = new GrpcAsyncRetryer();
 
   private final Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities;
 
@@ -58,12 +57,14 @@ public <T extends Throwable> void retry(RetryableProc<T> r, GrpcRetryerOptions o
 
   public <R, T extends Throwable> R retryWithResult(
       RetryableFunc<R, T> r, GrpcRetryerOptions options) throws T {
-    return SYNC.retry(r, options, serverCapabilities.get());
+    return new GrpcSyncRetryer().retry(r, options, serverCapabilities.get());
   }
 
   public <R> CompletableFuture<R> retryWithResultAsync(
-      Supplier<CompletableFuture<R>> function, GrpcRetryerOptions options) {
-    return ASYNC.retry(function, options, serverCapabilities.get());
+      ScheduledExecutorService executor,
+      Supplier<CompletableFuture<R>> function,
+      GrpcRetryerOptions options) {
+    return new GrpcAsyncRetryer<R>(executor, function, options, serverCapabilities.get()).retry();
   }
 
   public static class GrpcRetryerOptions {
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 1fb20f3045..e80444a3b2 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
@@ -57,7 +57,10 @@ public <R, T extends Throwable> R retry(
       attempt++;
 
       try {
-        throttler.throttle();
+        long throttleMs = throttler.getSleepTime();
+        if (throttleMs > 0) {
+          Thread.sleep(throttleMs);
+        }
         if (lastMeaningfulException != null) {
           log.debug("Retrying after failure", lastMeaningfulException);
         }
@@ -76,11 +79,10 @@ public <R, T extends Throwable> R retry(
         }
         lastMeaningfulException =
             GrpcRetryerUtils.lastMeaningfulException(e, lastMeaningfulException);
+        throttler.failure();
       }
       // 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.
-
-      throttler.failure();
     } while (!GrpcRetryerUtils.ranOutOfRetries(
         rpcOptions, attempt, retriesExpirationDeadline, Context.current().getDeadline()));
 
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 d04f2064cb..ffe43c8d9a 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
@@ -37,8 +37,6 @@
 
 public class GrpcAsyncRetryerTest {
 
-  private static final GrpcAsyncRetryer DEFAULT_ASYNC_RETRYER = new GrpcAsyncRetryer();
-
   private static ScheduledExecutorService scheduledExecutor;
 
   @BeforeClass
@@ -62,14 +60,17 @@ public void testExpirationAsync() throws InterruptedException {
             .setExpiration(Duration.ofMillis(500))
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
-      DEFAULT_ASYNC_RETRYER
-          .retry(
+      new GrpcAsyncRetryer<>(
+              scheduledExecutor,
               () -> {
+                attempts.incrementAndGet();
                 throw new StatusRuntimeException(Status.fromCode(STATUS_CODE));
               },
               new GrpcRetryer.GrpcRetryerOptions(options, null),
               GetSystemInfoResponse.Capabilities.getDefaultInstance())
+          .retry()
           .get();
       fail("unreachable");
     } catch (ExecutionException e) {
@@ -77,6 +78,7 @@ public void testExpirationAsync() throws InterruptedException {
       assertEquals(STATUS_CODE, ((StatusRuntimeException) e.getCause()).getStatus().getCode());
     }
 
+    assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1);
     assertTrue(System.currentTimeMillis() - start > 500);
   }
 
@@ -91,10 +93,12 @@ public void testExpirationFutureAsync() throws InterruptedException {
             .setExpiration(Duration.ofMillis(500))
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
-      DEFAULT_ASYNC_RETRYER
-          .retry(
+      new GrpcAsyncRetryer<>(
+              scheduledExecutor,
               () -> {
+                attempts.incrementAndGet();
                 CompletableFuture<Void> result = new CompletableFuture<>();
                 result.completeExceptionally(
                     new StatusRuntimeException(Status.fromCode(STATUS_CODE)));
@@ -102,12 +106,15 @@ public void testExpirationFutureAsync() throws InterruptedException {
               },
               new GrpcRetryer.GrpcRetryerOptions(options, null),
               GetSystemInfoResponse.Capabilities.getDefaultInstance())
+          .retry()
           .get();
       fail("unreachable");
     } catch (ExecutionException e) {
       assertTrue(e.getCause() instanceof StatusRuntimeException);
       assertEquals(STATUS_CODE, ((StatusRuntimeException) e.getCause()).getStatus().getCode());
     }
+
+    assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1);
     assertTrue(System.currentTimeMillis() - start > 500);
   }
 
@@ -117,15 +124,18 @@ public void testDoNotRetryAsync() throws InterruptedException {
 
     RpcRetryOptions options =
         RpcRetryOptions.newBuilder()
-            .setInitialInterval(Duration.ofMillis(10))
-            .setMaximumInterval(Duration.ofMillis(100))
+            .setInitialInterval(Duration.ofMillis(1000))
+            .setMaximumInterval(Duration.ofMillis(1000))
             .addDoNotRetry(STATUS_CODE, null)
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
-      DEFAULT_ASYNC_RETRYER
-          .retry(
+      new GrpcAsyncRetryer<>(
+              scheduledExecutor,
               () -> {
+                if (attempts.incrementAndGet() > 1)
+                  fail("We should not retry on exception that we specified to don't retry");
                 CompletableFuture<Void> result = new CompletableFuture<>();
                 result.completeExceptionally(
                     new StatusRuntimeException(Status.fromCode(STATUS_CODE)));
@@ -133,6 +143,7 @@ public void testDoNotRetryAsync() throws InterruptedException {
               },
               new GrpcRetryer.GrpcRetryerOptions(options, null),
               GetSystemInfoResponse.Capabilities.getDefaultInstance())
+          .retry()
           .get();
       fail("unreachable");
     } catch (ExecutionException e) {
@@ -141,27 +152,31 @@ public void testDoNotRetryAsync() throws InterruptedException {
     }
     assertTrue(
         "We should fail fast on exception that we specified to don't retry",
-        System.currentTimeMillis() - start < 10_000);
+        System.currentTimeMillis() - start < 1000);
   }
 
   @Test
   public void testInterruptedExceptionAsync() throws InterruptedException {
     RpcRetryOptions options =
         RpcRetryOptions.newBuilder()
-            .setInitialInterval(Duration.ofMillis(10))
-            .setMaximumInterval(Duration.ofMillis(100))
+            .setInitialInterval(Duration.ofMillis(1000))
+            .setMaximumInterval(Duration.ofMillis(1000))
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
-      DEFAULT_ASYNC_RETRYER
-          .retry(
+      new GrpcAsyncRetryer<>(
+              scheduledExecutor,
               () -> {
+                if (attempts.incrementAndGet() > 1)
+                  fail("We should not retry on InterruptedException");
                 CompletableFuture<Void> result = new CompletableFuture<>();
                 result.completeExceptionally(new InterruptedException("simulated"));
                 return result;
               },
               new GrpcRetryer.GrpcRetryerOptions(options, null),
               GetSystemInfoResponse.Capabilities.getDefaultInstance())
+          .retry()
           .get();
       fail("unreachable");
     } catch (ExecutionException e) {
@@ -169,27 +184,31 @@ public void testInterruptedExceptionAsync() throws InterruptedException {
       assertEquals("simulated", e.getCause().getMessage());
     }
     assertTrue(
-        "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 10_000);
+        "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 1000);
   }
 
   @Test
   public void testNotStatusRuntimeExceptionAsync() throws InterruptedException {
     RpcRetryOptions options =
         RpcRetryOptions.newBuilder()
-            .setInitialInterval(Duration.ofMillis(10))
-            .setMaximumInterval(Duration.ofMillis(100))
+            .setInitialInterval(Duration.ofMillis(1000))
+            .setMaximumInterval(Duration.ofMillis(1000))
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
-      DEFAULT_ASYNC_RETRYER
-          .retry(
+      new GrpcAsyncRetryer<>(
+              scheduledExecutor,
               () -> {
+                if (attempts.incrementAndGet() > 1)
+                  fail("We should not retry if the exception is not StatusRuntimeException");
                 CompletableFuture<Void> result = new CompletableFuture<>();
                 result.completeExceptionally(new IllegalArgumentException("simulated"));
                 return result;
               },
               new GrpcRetryer.GrpcRetryerOptions(options, null),
               GetSystemInfoResponse.Capabilities.getDefaultInstance())
+          .retry()
           .get();
       fail("unreachable");
     } catch (ExecutionException e) {
@@ -198,7 +217,7 @@ public void testNotStatusRuntimeExceptionAsync() throws InterruptedException {
     }
     assertTrue(
         "If the exception is not StatusRuntimeException - we shouldn't retry",
-        System.currentTimeMillis() - start < 10_000);
+        System.currentTimeMillis() - start < 1000);
   }
 
   @Test
@@ -217,8 +236,8 @@ public void testRetryDeadlineExceededException() {
             StatusRuntimeException.class,
             () -> {
               try {
-                DEFAULT_ASYNC_RETRYER
-                    .retry(
+                new GrpcAsyncRetryer<>(
+                        scheduledExecutor,
                         () -> {
                           attempts.incrementAndGet();
                           CompletableFuture<?> future = new CompletableFuture<>();
@@ -229,6 +248,7 @@ public void testRetryDeadlineExceededException() {
                         },
                         new GrpcRetryer.GrpcRetryerOptions(options, null),
                         GetSystemInfoResponse.Capabilities.getDefaultInstance())
+                    .retry()
                     .get();
               } catch (ExecutionException ex) {
                 throw ex.getCause();
@@ -239,7 +259,7 @@ public void testRetryDeadlineExceededException() {
 
     assertTrue(
         "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.",
-        System.currentTimeMillis() - start > 400);
+        System.currentTimeMillis() - start > 500);
 
     assertTrue(
         "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.",
@@ -267,8 +287,8 @@ public void testRespectGlobalDeadlineExceeded() {
                         StatusRuntimeException.class,
                         () -> {
                           try {
-                            DEFAULT_ASYNC_RETRYER
-                                .retry(
+                            new GrpcAsyncRetryer<>(
+                                    scheduledExecutor,
                                     () -> {
                                       attempts.incrementAndGet();
                                       CompletableFuture<?> future = new CompletableFuture<>();
@@ -279,6 +299,7 @@ public void testRespectGlobalDeadlineExceeded() {
                                     },
                                     new GrpcRetryer.GrpcRetryerOptions(options, null),
                                     GetSystemInfoResponse.Capabilities.getDefaultInstance())
+                                .retry()
                                 .get();
                           } catch (ExecutionException e) {
                             throw e.getCause();
@@ -313,8 +334,8 @@ public void testGlobalDeadlineExceededAfterAnotherException() {
                         StatusRuntimeException.class,
                         () -> {
                           try {
-                            DEFAULT_ASYNC_RETRYER
-                                .retry(
+                            new GrpcAsyncRetryer<>(
+                                    scheduledExecutor,
                                     () -> {
                                       if (Context.current().getDeadline().isExpired()) {
                                         throw new StatusRuntimeException(
@@ -326,6 +347,7 @@ public void testGlobalDeadlineExceededAfterAnotherException() {
                                     },
                                     new GrpcRetryer.GrpcRetryerOptions(options, null),
                                     GetSystemInfoResponse.Capabilities.getDefaultInstance())
+                                .retry()
                                 .get();
                           } catch (ExecutionException e) {
                             throw e.getCause();
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 54ca31766b..3124dd7c5b 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
@@ -65,9 +65,11 @@ public void testExpiration() {
             .setExpiration(Duration.ofMillis(500))
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
       DEFAULT_SYNC_RETRYER.retry(
           () -> {
+            attempts.incrementAndGet();
             throw new StatusRuntimeException(Status.fromCode(STATUS_CODE));
           },
           new GrpcRetryer.GrpcRetryerOptions(options, null),
@@ -78,7 +80,8 @@ public void testExpiration() {
       assertEquals(STATUS_CODE, ((StatusRuntimeException) e).getStatus().getCode());
     }
 
-    assertTrue(System.currentTimeMillis() - start > 400);
+    assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1);
+    assertTrue(System.currentTimeMillis() - start > 500);
   }
 
   @Test
@@ -87,14 +90,17 @@ public void testDoNotRetry() {
 
     RpcRetryOptions options =
         RpcRetryOptions.newBuilder()
-            .setInitialInterval(Duration.ofMillis(10))
-            .setMaximumInterval(Duration.ofMillis(100))
+            .setInitialInterval(Duration.ofMillis(1000))
+            .setMaximumInterval(Duration.ofMillis(1000))
             .addDoNotRetry(STATUS_CODE, null)
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
       DEFAULT_SYNC_RETRYER.retry(
           () -> {
+            if (attempts.incrementAndGet() > 1)
+              fail("We should not retry on exception that we specified to don't retry");
             throw new StatusRuntimeException(Status.fromCode(STATUS_CODE));
           },
           new GrpcRetryer.GrpcRetryerOptions(options, null),
@@ -106,20 +112,22 @@ public void testDoNotRetry() {
     }
     assertTrue(
         "We should fail fast on exception that we specified to don't retry",
-        System.currentTimeMillis() - start < 10_000);
+        System.currentTimeMillis() - start < 1000);
   }
 
   @Test
   public void testInterruptedException() {
     RpcRetryOptions options =
         RpcRetryOptions.newBuilder()
-            .setInitialInterval(Duration.ofMillis(10))
-            .setMaximumInterval(Duration.ofMillis(100))
+            .setInitialInterval(Duration.ofMillis(1000))
+            .setMaximumInterval(Duration.ofMillis(1000))
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
       DEFAULT_SYNC_RETRYER.retry(
           () -> {
+            if (attempts.incrementAndGet() > 1) fail("We should not retry on InterruptedException");
             throw new InterruptedException();
           },
           new GrpcRetryer.GrpcRetryerOptions(options, null),
@@ -129,20 +137,23 @@ public void testInterruptedException() {
       assertTrue(e instanceof CancellationException);
     }
     assertTrue(
-        "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 10_000);
+        "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 1000);
   }
 
   @Test
   public void testNotStatusRuntimeException() {
     RpcRetryOptions options =
         RpcRetryOptions.newBuilder()
-            .setInitialInterval(Duration.ofMillis(10))
-            .setMaximumInterval(Duration.ofMillis(100))
+            .setInitialInterval(Duration.ofMillis(1000))
+            .setMaximumInterval(Duration.ofMillis(1000))
             .validateBuildWithDefaults();
     long start = System.currentTimeMillis();
+    final AtomicInteger attempts = new AtomicInteger();
     try {
       DEFAULT_SYNC_RETRYER.retry(
           () -> {
+            if (attempts.incrementAndGet() > 1)
+              fail("We should not retry if the exception is not StatusRuntimeException");
             throw new IllegalArgumentException("simulated");
           },
           new GrpcRetryer.GrpcRetryerOptions(options, null),
@@ -153,8 +164,8 @@ public void testNotStatusRuntimeException() {
       assertEquals("simulated", e.getMessage());
     }
     assertTrue(
-        "If the exception is not StatusRuntimeException - we shouldn't retry",
-        System.currentTimeMillis() - start < 10_000);
+        "Should fail fast if the exception is not StatusRuntimeException",
+        System.currentTimeMillis() - start < 1000);
   }
 
   @Test
@@ -184,7 +195,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 > 400);
+        System.currentTimeMillis() - start > 500);
 
     assertTrue(
         "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.",

From 3d0026e0b50d1eb518119df86e154b8fa4c0752f Mon Sep 17 00:00:00 2001
From: James Watkins-Harvey <mjameswh@gmail.com>
Date: Wed, 12 Oct 2022 16:34:41 -0400
Subject: [PATCH 4/4] Add @NotThreadSafe

---
 .../src/main/java/io/temporal/internal/BackoffThrottler.java    | 2 ++
 1 file changed, 2 insertions(+)

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 5e741cd09d..2e59513055 100644
--- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
+++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java
@@ -23,6 +23,7 @@
 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
@@ -58,6 +59,7 @@
  *
  * @author fateev
  */
+@NotThreadSafe
 public final class BackoffThrottler {
 
   private final Duration initialSleep;