diff --git a/gax/src/main/java/com/google/api/gax/retrying/BasicRetryingFuture.java b/gax/src/main/java/com/google/api/gax/retrying/BasicRetryingFuture.java
index ce360e3b8..331a1e304 100644
--- a/gax/src/main/java/com/google/api/gax/retrying/BasicRetryingFuture.java
+++ b/gax/src/main/java/com/google/api/gax/retrying/BasicRetryingFuture.java
@@ -57,15 +57,20 @@ class BasicRetryingFuture<ResponseT> extends AbstractFuture<ResponseT>
   private final Callable<ResponseT> callable;
 
   private final RetryAlgorithm<ResponseT> retryAlgorithm;
+  private final RetryingContext retryingContext;
 
   private volatile TimedAttemptSettings attemptSettings;
 
   private volatile ApiFuture<ResponseT> latestCompletedAttemptResult;
   private volatile ApiFuture<ResponseT> attemptResult;
 
-  BasicRetryingFuture(Callable<ResponseT> callable, RetryAlgorithm<ResponseT> retryAlgorithm) {
+  BasicRetryingFuture(
+      Callable<ResponseT> callable,
+      RetryAlgorithm<ResponseT> retryAlgorithm,
+      RetryingContext context) {
     this.callable = checkNotNull(callable);
     this.retryAlgorithm = checkNotNull(retryAlgorithm);
+    this.retryingContext = checkNotNull(context);
 
     this.attemptSettings = retryAlgorithm.createFirstAttempt();
 
diff --git a/gax/src/main/java/com/google/api/gax/retrying/CallbackChainRetryingFuture.java b/gax/src/main/java/com/google/api/gax/retrying/CallbackChainRetryingFuture.java
index b65f4d296..f6714e23b 100644
--- a/gax/src/main/java/com/google/api/gax/retrying/CallbackChainRetryingFuture.java
+++ b/gax/src/main/java/com/google/api/gax/retrying/CallbackChainRetryingFuture.java
@@ -52,14 +52,15 @@
  * <p>This class is thread-safe.
  */
 class CallbackChainRetryingFuture<ResponseT> extends BasicRetryingFuture<ResponseT> {
-  private final RetryingExecutor<ResponseT> retryingExecutor;
+  private final ScheduledRetryingExecutor<ResponseT> retryingExecutor;
   private volatile AttemptCompletionListener attemptFutureCompletionListener;
 
   CallbackChainRetryingFuture(
       Callable<ResponseT> callable,
       RetryAlgorithm<ResponseT> retryAlgorithm,
-      RetryingExecutor<ResponseT> retryingExecutor) {
-    super(callable, retryAlgorithm);
+      ScheduledRetryingExecutor<ResponseT> retryingExecutor,
+      RetryingContext context) {
+    super(callable, retryAlgorithm, context);
     this.retryingExecutor = checkNotNull(retryingExecutor);
   }
 
diff --git a/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java b/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java
index 361ad1462..c024bda77 100644
--- a/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java
+++ b/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java
@@ -33,6 +33,7 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
 import java.io.InterruptedIOException;
 import java.nio.channels.ClosedByInterruptException;
 import java.util.concurrent.Callable;
@@ -46,7 +47,7 @@
  *
  * @param <ResponseT> response type
  */
-public class DirectRetryingExecutor<ResponseT> implements RetryingExecutor<ResponseT> {
+public class DirectRetryingExecutor<ResponseT> implements RetryingExecutorWithContext<ResponseT> {
 
   private final RetryAlgorithm<ResponseT> retryAlgorithm;
 
@@ -70,7 +71,21 @@ public DirectRetryingExecutor(RetryAlgorithm<ResponseT> retryAlgorithm) {
    */
   @Override
   public RetryingFuture<ResponseT> createFuture(Callable<ResponseT> callable) {
-    return new BasicRetryingFuture<>(callable, retryAlgorithm);
+    return createFuture(callable, NoopRetryingContext.create());
+  }
+
+  /**
+   * Creates a {@link RetryingFuture}, which is a facade, returned to the client code to wait for
+   * any retriable operation to complete. The future is bounded to {@code this} executor instance.
+   *
+   * @param callable the actual callable, which should be executed in a retriable context
+   * @return retrying future facade
+   */
+  @BetaApi("The surface for passing per operation state is not yet stable")
+  @Override
+  public RetryingFuture<ResponseT> createFuture(
+      Callable<ResponseT> callable, RetryingContext context) {
+    return new BasicRetryingFuture<>(callable, retryAlgorithm, context);
   }
 
   /**
@@ -78,7 +93,8 @@ public RetryingFuture<ResponseT> createFuture(Callable<ResponseT> callable) {
    * the specified by the {@link RetryingFuture#getAttemptSettings()} amount of time. As result,
    * this method completes execution only after the specified {@code retryingFuture} completes.
    *
-   * @param retryingFuture the future previously returned by {@link #createFuture(Callable)}
+   * @param retryingFuture the future previously returned by {@link #createFuture(Callable,
+   *     RetryingContext)}
    * @return returns completed {@code retryingFuture}
    */
   @Override
diff --git a/gax/src/main/java/com/google/api/gax/retrying/NoopRetryingContext.java b/gax/src/main/java/com/google/api/gax/retrying/NoopRetryingContext.java
new file mode 100644
index 000000000..def9cc796
--- /dev/null
+++ b/gax/src/main/java/com/google/api/gax/retrying/NoopRetryingContext.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.retrying;
+
+// TODO(igorbernstein2): Remove this class once RetryingExecutor#createFuture(Callable) is
+// deprecated and removed.
+/**
+ * Backwards compatibility class to aid in transition to adding operation state to {@link
+ * RetryingFuture} implementations.
+ */
+class NoopRetryingContext implements RetryingContext {
+  public static RetryingContext create() {
+    return new NoopRetryingContext();
+  }
+}
diff --git a/gax/src/main/java/com/google/api/gax/retrying/RetryingContext.java b/gax/src/main/java/com/google/api/gax/retrying/RetryingContext.java
new file mode 100644
index 000000000..335d14611
--- /dev/null
+++ b/gax/src/main/java/com/google/api/gax/retrying/RetryingContext.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.retrying;
+
+import com.google.api.core.BetaApi;
+
+/**
+ * Context for a retryable operation.
+ *
+ * <p>It provides state to individual {@link RetryingFuture}s via the {@link RetryingExecutor}.
+ */
+@BetaApi("The surface for passing per operation state is not yet stable")
+public interface RetryingContext {}
diff --git a/gax/src/main/java/com/google/api/gax/retrying/RetryingExecutorWithContext.java b/gax/src/main/java/com/google/api/gax/retrying/RetryingExecutorWithContext.java
new file mode 100644
index 000000000..f19ba5607
--- /dev/null
+++ b/gax/src/main/java/com/google/api/gax/retrying/RetryingExecutorWithContext.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google LLC nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.google.api.gax.retrying;
+
+import com.google.api.core.BetaApi;
+import com.google.api.core.InternalExtensionOnly;
+import java.util.concurrent.Callable;
+import javax.annotation.Nonnull;
+
+/**
+ * A {@link RetryingExecutor} that accepts a per-operation context.
+ *
+ * @see RetryingExecutor
+ */
+// TODO(igorbernstein2): Consider replacing this with a default implementation in RetryingExecutor
+// once support for java 7 is dropped
+@BetaApi("The surface for per invocation state is unstable and will probably change in the future")
+@InternalExtensionOnly
+public interface RetryingExecutorWithContext<ResponseT> extends RetryingExecutor<ResponseT> {
+  RetryingFuture<ResponseT> createFuture(
+      @Nonnull Callable<ResponseT> callable, @Nonnull RetryingContext context);
+}
diff --git a/gax/src/main/java/com/google/api/gax/retrying/ScheduledRetryingExecutor.java b/gax/src/main/java/com/google/api/gax/retrying/ScheduledRetryingExecutor.java
index c6bb8d855..b6ee0a0c1 100644
--- a/gax/src/main/java/com/google/api/gax/retrying/ScheduledRetryingExecutor.java
+++ b/gax/src/main/java/com/google/api/gax/retrying/ScheduledRetryingExecutor.java
@@ -31,6 +31,7 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
 import com.google.api.core.ListenableFutureToApiFuture;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningScheduledExecutorService;
@@ -53,7 +54,8 @@
  *
  * @param <ResponseT> response type
  */
-public class ScheduledRetryingExecutor<ResponseT> implements RetryingExecutor<ResponseT> {
+public class ScheduledRetryingExecutor<ResponseT>
+    implements RetryingExecutorWithContext<ResponseT> {
 
   private final RetryAlgorithm<ResponseT> retryAlgorithm;
   private final ListeningScheduledExecutorService scheduler;
@@ -81,13 +83,30 @@ public ScheduledRetryingExecutor(
    */
   @Override
   public RetryingFuture<ResponseT> createFuture(Callable<ResponseT> callable) {
-    return new CallbackChainRetryingFuture<>(callable, retryAlgorithm, this);
+    return createFuture(callable, NoopRetryingContext.create());
+  }
+
+  /**
+   * Creates a {@link RetryingFuture}, which is a facade, returned to the client code to wait for
+   * any retriable operation to complete. The returned future is bounded to {@code this} executor
+   * instance.
+   *
+   * @param callable the actual callable, which should be executed in a retriable context
+   * @param context the context for this operation
+   * @return retrying future facade
+   */
+  @BetaApi("The surface for passing per operation state is not yet stable")
+  @Override
+  public RetryingFuture<ResponseT> createFuture(
+      Callable<ResponseT> callable, RetryingContext context) {
+    return new CallbackChainRetryingFuture<>(callable, retryAlgorithm, this, context);
   }
 
   /**
    * Submits an attempt for execution in a different thread.
    *
-   * @param retryingFuture the future previously returned by {@link #createFuture(Callable)}
+   * @param retryingFuture the future previously returned by {@link #createFuture(Callable,
+   *     RetryingContext)}
    * @return submitted attempt future
    */
   @Override
diff --git a/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java b/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java
index f945319bf..658453ef7 100644
--- a/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java
+++ b/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java
@@ -31,6 +31,7 @@
 
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalExtensionOnly;
+import com.google.api.gax.retrying.RetryingContext;
 import com.google.auth.Credentials;
 import java.util.List;
 import java.util.Map;
@@ -48,7 +49,7 @@
  * <p>This is transport specific and each transport has an implementation with its own options.
  */
 @InternalExtensionOnly
-public interface ApiCallContext {
+public interface ApiCallContext extends RetryingContext {
 
   /** Returns a new ApiCallContext with the given credentials set. */
   ApiCallContext withCredentials(Credentials credentials);
diff --git a/gax/src/main/java/com/google/api/gax/rpc/Callables.java b/gax/src/main/java/com/google/api/gax/rpc/Callables.java
index e684c6dad..85b0f1af9 100644
--- a/gax/src/main/java/com/google/api/gax/rpc/Callables.java
+++ b/gax/src/main/java/com/google/api/gax/rpc/Callables.java
@@ -35,7 +35,6 @@
 import com.google.api.gax.longrunning.OperationSnapshot;
 import com.google.api.gax.retrying.ExponentialRetryAlgorithm;
 import com.google.api.gax.retrying.RetryAlgorithm;
-import com.google.api.gax.retrying.RetryingExecutor;
 import com.google.api.gax.retrying.ScheduledRetryingExecutor;
 import com.google.api.gax.retrying.StreamingRetryAlgorithm;
 
@@ -64,7 +63,7 @@ public static <RequestT, ResponseT> UnaryCallable<RequestT, ResponseT> retrying(
             new ApiResultRetryAlgorithm<ResponseT>(),
             new ExponentialRetryAlgorithm(
                 callSettings.getRetrySettings(), clientContext.getClock()));
-    RetryingExecutor<ResponseT> retryingExecutor =
+    ScheduledRetryingExecutor<ResponseT> retryingExecutor =
         new ScheduledRetryingExecutor<>(retryAlgorithm, clientContext.getExecutor());
     return new RetryingCallable<>(
         clientContext.getDefaultCallContext(), innerCallable, retryingExecutor);
diff --git a/gax/src/main/java/com/google/api/gax/rpc/OperationCallableImpl.java b/gax/src/main/java/com/google/api/gax/rpc/OperationCallableImpl.java
index aa9eee9be..f1161851b 100644
--- a/gax/src/main/java/com/google/api/gax/rpc/OperationCallableImpl.java
+++ b/gax/src/main/java/com/google/api/gax/rpc/OperationCallableImpl.java
@@ -36,7 +36,7 @@
 import com.google.api.gax.longrunning.OperationFuture;
 import com.google.api.gax.longrunning.OperationFutureImpl;
 import com.google.api.gax.longrunning.OperationSnapshot;
-import com.google.api.gax.retrying.RetryingExecutor;
+import com.google.api.gax.retrying.RetryingExecutorWithContext;
 import com.google.api.gax.retrying.RetryingFuture;
 
 /**
@@ -50,14 +50,14 @@ class OperationCallableImpl<RequestT, ResponseT, MetadataT>
     extends OperationCallable<RequestT, ResponseT, MetadataT> {
 
   private final UnaryCallable<RequestT, OperationSnapshot> initialCallable;
-  private final RetryingExecutor<OperationSnapshot> executor;
+  private final RetryingExecutorWithContext<OperationSnapshot> executor;
   private final LongRunningClient longRunningClient;
   private final ApiFunction<OperationSnapshot, ResponseT> responseTransformer;
   private final ApiFunction<OperationSnapshot, MetadataT> metadataTransformer;
 
   OperationCallableImpl(
       UnaryCallable<RequestT, OperationSnapshot> initialCallable,
-      RetryingExecutor<OperationSnapshot> executor,
+      RetryingExecutorWithContext<OperationSnapshot> executor,
       LongRunningClient longRunningClient,
       OperationCallSettings<RequestT, ResponseT, MetadataT> operationCallSettings) {
     this.initialCallable = checkNotNull(initialCallable);
diff --git a/gax/src/main/java/com/google/api/gax/rpc/RecheckingCallable.java b/gax/src/main/java/com/google/api/gax/rpc/RecheckingCallable.java
index 5f75323fb..9441223e4 100644
--- a/gax/src/main/java/com/google/api/gax/rpc/RecheckingCallable.java
+++ b/gax/src/main/java/com/google/api/gax/rpc/RecheckingCallable.java
@@ -29,7 +29,7 @@
  */
 package com.google.api.gax.rpc;
 
-import com.google.api.gax.retrying.RetryingExecutor;
+import com.google.api.gax.retrying.RetryingExecutorWithContext;
 import com.google.api.gax.retrying.RetryingFuture;
 import com.google.common.base.Preconditions;
 
@@ -43,10 +43,11 @@
  */
 class RecheckingCallable<RequestT, ResponseT> extends UnaryCallable<RequestT, ResponseT> {
   private final UnaryCallable<RequestT, ResponseT> callable;
-  private final RetryingExecutor<ResponseT> executor;
+  private final RetryingExecutorWithContext<ResponseT> executor;
 
   RecheckingCallable(
-      UnaryCallable<RequestT, ResponseT> callable, RetryingExecutor<ResponseT> executor) {
+      UnaryCallable<RequestT, ResponseT> callable,
+      RetryingExecutorWithContext<ResponseT> executor) {
     this.callable = Preconditions.checkNotNull(callable);
     this.executor = Preconditions.checkNotNull(executor);
   }
@@ -56,7 +57,8 @@ public RetryingFuture<ResponseT> futureCall(RequestT ignored, ApiCallContext inp
     CheckingAttemptCallable<RequestT, ResponseT> checkingAttemptCallable =
         new CheckingAttemptCallable<>(callable, inputContext);
 
-    RetryingFuture<ResponseT> retryingFuture = executor.createFuture(checkingAttemptCallable);
+    RetryingFuture<ResponseT> retryingFuture =
+        executor.createFuture(checkingAttemptCallable, inputContext);
     checkingAttemptCallable.setExternalFuture(retryingFuture);
     checkingAttemptCallable.call();
 
diff --git a/gax/src/main/java/com/google/api/gax/rpc/RetryingCallable.java b/gax/src/main/java/com/google/api/gax/rpc/RetryingCallable.java
index ba2307932..0a92794a2 100644
--- a/gax/src/main/java/com/google/api/gax/rpc/RetryingCallable.java
+++ b/gax/src/main/java/com/google/api/gax/rpc/RetryingCallable.java
@@ -29,7 +29,7 @@
  */
 package com.google.api.gax.rpc;
 
-import com.google.api.gax.retrying.RetryingExecutor;
+import com.google.api.gax.retrying.RetryingExecutorWithContext;
 import com.google.api.gax.retrying.RetryingFuture;
 import com.google.common.base.Preconditions;
 
@@ -41,12 +41,12 @@
 class RetryingCallable<RequestT, ResponseT> extends UnaryCallable<RequestT, ResponseT> {
   private final ApiCallContext callContextPrototype;
   private final UnaryCallable<RequestT, ResponseT> callable;
-  private final RetryingExecutor<ResponseT> executor;
+  private final RetryingExecutorWithContext<ResponseT> executor;
 
   RetryingCallable(
       ApiCallContext callContextPrototype,
       UnaryCallable<RequestT, ResponseT> callable,
-      RetryingExecutor<ResponseT> executor) {
+      RetryingExecutorWithContext<ResponseT> executor) {
     this.callContextPrototype = Preconditions.checkNotNull(callContextPrototype);
     this.callable = Preconditions.checkNotNull(callable);
     this.executor = Preconditions.checkNotNull(executor);
@@ -58,7 +58,7 @@ public RetryingFuture<ResponseT> futureCall(RequestT request, ApiCallContext inp
     AttemptCallable<RequestT, ResponseT> retryCallable =
         new AttemptCallable<>(callable, request, context);
 
-    RetryingFuture<ResponseT> retryingFuture = executor.createFuture(retryCallable);
+    RetryingFuture<ResponseT> retryingFuture = executor.createFuture(retryCallable, inputContext);
     retryCallable.setExternalFuture(retryingFuture);
     retryCallable.call();
 
diff --git a/gax/src/main/java/com/google/api/gax/rpc/RetryingServerStreamingCallable.java b/gax/src/main/java/com/google/api/gax/rpc/RetryingServerStreamingCallable.java
index 3c4aec370..cb7e3bb78 100644
--- a/gax/src/main/java/com/google/api/gax/rpc/RetryingServerStreamingCallable.java
+++ b/gax/src/main/java/com/google/api/gax/rpc/RetryingServerStreamingCallable.java
@@ -82,7 +82,7 @@ public void call(
             context,
             responseObserver);
 
-    RetryingFuture<Void> retryingFuture = executor.createFuture(attemptCallable);
+    RetryingFuture<Void> retryingFuture = executor.createFuture(attemptCallable, context);
     attemptCallable.setExternalFuture(retryingFuture);
     attemptCallable.start();
 
diff --git a/gax/src/test/java/com/google/api/gax/retrying/AbstractRetryingExecutorTest.java b/gax/src/test/java/com/google/api/gax/retrying/AbstractRetryingExecutorTest.java
index 167774e02..08f6562eb 100644
--- a/gax/src/test/java/com/google/api/gax/retrying/AbstractRetryingExecutorTest.java
+++ b/gax/src/test/java/com/google/api/gax/retrying/AbstractRetryingExecutorTest.java
@@ -47,12 +47,16 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
 import org.threeten.bp.Duration;
 
-@RunWith(JUnit4.class)
+@RunWith(MockitoJUnitRunner.class)
 public abstract class AbstractRetryingExecutorTest {
-  protected abstract RetryingExecutor<String> getExecutor(RetryAlgorithm<String> retryAlgorithm);
+  @Mock protected RetryingContext retryingContext;
+
+  protected abstract RetryingExecutorWithContext<String> getExecutor(
+      RetryAlgorithm<String> retryAlgorithm);
 
   protected abstract RetryAlgorithm<String> getAlgorithm(
       RetrySettings retrySettings, int apocalypseCountDown, RuntimeException apocalypseException);
@@ -60,8 +64,9 @@ protected abstract RetryAlgorithm<String> getAlgorithm(
   @Test
   public void testSuccess() throws Exception {
     FailingCallable callable = new FailingCallable(0, "SUCCESS");
-    RetryingExecutor<String> executor = getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingExecutorWithContext<String> executor =
+        getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     future.setAttemptFuture(executor.submit(future));
 
     assertFutureSuccess(future);
@@ -71,8 +76,9 @@ public void testSuccess() throws Exception {
   @Test
   public void testSuccessWithFailures() throws Exception {
     FailingCallable callable = new FailingCallable(5, "SUCCESS");
-    RetryingExecutor<String> executor = getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingExecutorWithContext<String> executor =
+        getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     future.setAttemptFuture(executor.submit(future));
 
     assertFutureSuccess(future);
@@ -82,8 +88,9 @@ public void testSuccessWithFailures() throws Exception {
   @Test
   public void testSuccessWithFailuresPeekGetAttempt() throws Exception {
     FailingCallable callable = new FailingCallable(5, "SUCCESS");
-    RetryingExecutor<String> executor = getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingExecutorWithContext<String> executor =
+        getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
 
     assertNull(future.peekAttemptResult());
     assertSame(future.peekAttemptResult(), future.peekAttemptResult());
@@ -107,8 +114,9 @@ public void testSuccessWithFailuresPeekGetAttempt() throws Exception {
   @Test
   public void testMaxRetriesExceeded() throws Exception {
     FailingCallable callable = new FailingCallable(6, "FAILURE");
-    RetryingExecutor<String> executor = getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingExecutorWithContext<String> executor =
+        getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 0, null));
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     future.setAttemptFuture(executor.submit(future));
 
     assertFutureFail(future, CustomException.class);
@@ -123,9 +131,10 @@ public void testTotalTimeoutExceeded() throws Exception {
             .setInitialRetryDelay(Duration.ofMillis(Integer.MAX_VALUE))
             .setMaxRetryDelay(Duration.ofMillis(Integer.MAX_VALUE))
             .build();
-    RetryingExecutor<String> executor = getExecutor(getAlgorithm(retrySettings, 0, null));
+    RetryingExecutorWithContext<String> executor =
+        getExecutor(getAlgorithm(retrySettings, 0, null));
     FailingCallable callable = new FailingCallable(6, "FAILURE");
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     future.setAttemptFuture(executor.submit(future));
 
     assertFutureFail(future, CustomException.class);
@@ -143,8 +152,9 @@ public void testCancelOuterFutureBeforeStart() throws Exception {
             .setMaxRetryDelay(Duration.ofMillis(1_000L))
             .setTotalTimeout(Duration.ofMillis(10_0000L))
             .build();
-    RetryingExecutor<String> executor = getExecutor(getAlgorithm(retrySettings, 0, null));
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingExecutorWithContext<String> executor =
+        getExecutor(getAlgorithm(retrySettings, 0, null));
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     boolean res = future.cancel(false);
 
     assertTrue(res);
@@ -158,9 +168,9 @@ public void testCancelOuterFutureBeforeStart() throws Exception {
   @Test
   public void testCancelByRetryingAlgorithm() throws Exception {
     FailingCallable callable = new FailingCallable(6, "FAILURE");
-    RetryingExecutor<String> executor =
+    RetryingExecutorWithContext<String> executor =
         getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 5, new CancellationException()));
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     future.setAttemptFuture(executor.submit(future));
 
     assertFutureCancel(future);
@@ -170,9 +180,9 @@ public void testCancelByRetryingAlgorithm() throws Exception {
   @Test
   public void testUnexpectedExceptionFromRetryAlgorithm() throws Exception {
     FailingCallable callable = new FailingCallable(6, "FAILURE");
-    RetryingExecutor<String> executor =
+    RetryingExecutorWithContext<String> executor =
         getExecutor(getAlgorithm(FAST_RETRY_SETTINGS, 5, new RuntimeException()));
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     future.setAttemptFuture(executor.submit(future));
 
     assertFutureFail(future, RuntimeException.class);
@@ -193,9 +203,9 @@ public void testPollExceptionByPollAlgorithm() throws Exception {
             new TestResultRetryAlgorithm<String>(0, null),
             new ExponentialPollAlgorithm(retrySettings, NanoClock.getDefaultClock()));
 
-    RetryingExecutor<String> executor = getExecutor(retryAlgorithm);
+    RetryingExecutorWithContext<String> executor = getExecutor(retryAlgorithm);
     FailingCallable callable = new FailingCallable(6, "FAILURE");
-    RetryingFuture<String> future = executor.createFuture(callable);
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
     future.setAttemptFuture(executor.submit(future));
 
     assertFutureFail(future, PollException.class);
diff --git a/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java b/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java
index 03208cb74..b8a990fbb 100644
--- a/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java
+++ b/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java
@@ -31,13 +31,13 @@
 
 import com.google.api.core.CurrentMillisClock;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.mockito.junit.MockitoJUnitRunner;
 
-@RunWith(JUnit4.class)
+@RunWith(MockitoJUnitRunner.class)
 public class DirectRetryingExecutorTest extends AbstractRetryingExecutorTest {
 
   @Override
-  protected RetryingExecutor<String> getExecutor(RetryAlgorithm<String> retryAlgorithm) {
+  protected RetryingExecutorWithContext<String> getExecutor(RetryAlgorithm<String> retryAlgorithm) {
     return new DirectRetryingExecutor<>(retryAlgorithm);
   }
 
diff --git a/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java b/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java
index 0aff90d01..2d4708871 100644
--- a/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java
+++ b/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java
@@ -49,10 +49,10 @@
 import org.junit.After;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.mockito.junit.MockitoJUnitRunner;
 import org.threeten.bp.Duration;
 
-@RunWith(JUnit4.class)
+@RunWith(MockitoJUnitRunner.class)
 public class ScheduledRetryingExecutorTest extends AbstractRetryingExecutorTest {
   private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
 
@@ -60,7 +60,7 @@ public class ScheduledRetryingExecutorTest extends AbstractRetryingExecutorTest
   private static final int EXECUTIONS_COUNT = 5;
 
   @Override
-  protected RetryingExecutor<String> getExecutor(RetryAlgorithm<String> retryAlgorithm) {
+  protected RetryingExecutorWithContext<String> getExecutor(RetryAlgorithm<String> retryAlgorithm) {
     return getRetryingExecutor(retryAlgorithm, scheduler);
   }
 
@@ -72,7 +72,7 @@ protected RetryAlgorithm<String> getAlgorithm(
         new ExponentialRetryAlgorithm(retrySettings, NanoClock.getDefaultClock()));
   }
 
-  private RetryingExecutor<String> getRetryingExecutor(
+  private RetryingExecutorWithContext<String> getRetryingExecutor(
       RetryAlgorithm<String> retryAlgorithm, ScheduledExecutorService scheduler) {
 
     return new ScheduledRetryingExecutor<>(retryAlgorithm, scheduler);
@@ -98,9 +98,9 @@ public void testSuccessWithFailuresPeekAttempt() throws Exception {
               .setMaxAttempts(maxRetries)
               .build();
 
-      RetryingExecutor<String> executor =
+      RetryingExecutorWithContext<String> executor =
           getRetryingExecutor(getAlgorithm(retrySettings, 0, null), localExecutor);
-      RetryingFuture<String> future = executor.createFuture(callable);
+      RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
 
       assertNull(future.peekAttemptResult());
       assertSame(future.peekAttemptResult(), future.peekAttemptResult());
@@ -147,9 +147,9 @@ public void testSuccessWithFailuresGetAttempt() throws Exception {
               .setMaxAttempts(maxRetries)
               .build();
 
-      RetryingExecutor<String> executor =
+      RetryingExecutorWithContext<String> executor =
           getRetryingExecutor(getAlgorithm(retrySettings, 0, null), localExecutor);
-      RetryingFuture<String> future = executor.createFuture(callable);
+      RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
 
       assertNull(future.peekAttemptResult());
       assertSame(future.getAttemptResult(), future.getAttemptResult());
@@ -199,9 +199,9 @@ public void testCancelGetAttempt() throws Exception {
               .setMaxAttempts(maxRetries)
               .build();
 
-      RetryingExecutor<String> executor =
+      RetryingExecutorWithContext<String> executor =
           getRetryingExecutor(getAlgorithm(retrySettings, 0, null), localExecutor);
-      RetryingFuture<String> future = executor.createFuture(callable);
+      RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
 
       assertNull(future.peekAttemptResult());
       assertSame(future.getAttemptResult(), future.getAttemptResult());
@@ -255,9 +255,9 @@ public void testCancelOuterFutureAfterStart() throws Exception {
               .setMaxRetryDelay(Duration.ofMillis(1_000L))
               .setTotalTimeout(Duration.ofMillis(10_0000L))
               .build();
-      RetryingExecutor<String> executor =
+      RetryingExecutorWithContext<String> executor =
           getRetryingExecutor(getAlgorithm(retrySettings, 0, null), localExecutor);
-      RetryingFuture<String> future = executor.createFuture(callable);
+      RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
       future.setAttemptFuture(executor.submit(future));
 
       Thread.sleep(30L);
@@ -283,9 +283,9 @@ public void testCancelProxiedFutureAfterStart() throws Exception {
               .setMaxRetryDelay(Duration.ofMillis(1_000L))
               .setTotalTimeout(Duration.ofMillis(10_0000L))
               .build();
-      RetryingExecutor<String> executor =
+      RetryingExecutorWithContext<String> executor =
           getRetryingExecutor(getAlgorithm(retrySettings, 0, null), localExecutor);
-      RetryingFuture<String> future = executor.createFuture(callable);
+      RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
       future.setAttemptFuture(executor.submit(future));
 
       Thread.sleep(50L);