Repository: googleapis/java-bigtable-hbase
Pull request #2717 (2020-11-25 20:54:13)

feat: Adding timeouts for RPC Retry Attempts for HBase client (on bigtable-1.x branch)
We'd like to give user the ability to set a shorter timeout for individual RPC attempts within an overall timeout window (e.g. to allow multiple retries during the timeout window)


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/pom.xml b/bigtable-client-core-parent/bigtable-client-core/pom.xml
index c07f2980e..7a2740f16 100644
--- a/bigtable-client-core-parent/bigtable-client-core/pom.xml
+++ b/bigtable-client-core-parent/bigtable-client-core/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
   </parent>
 
   <artifactId>bigtable-client-core</artifactId>
@@ -247,6 +247,12 @@ this is not used directly, but upgrading due to transitive vulnerabilities in ol
       <version>${hamcrest.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-library</artifactId>
+      <version>${hamcrest.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/CallOptionsConfig.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/CallOptionsConfig.java
index 4d927777a..bb6c26782 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/CallOptionsConfig.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/CallOptionsConfig.java
@@ -15,14 +15,18 @@
  */
 package com.google.cloud.bigtable.config;
 
+import com.google.api.core.BetaApi;
 import com.google.api.core.InternalExtensionOnly;
 import com.google.common.base.MoreObjects;
+import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import io.grpc.CallOptions;
 import java.io.Serializable;
+import java.util.Objects;
 
 /**
- * Experimental options to turn on timeout options. {@link io.grpc.CallOptions} supports other
- * settings as well, which this configuration object could help set.
+ * Experimental options to turn on timeout options. {@link CallOptions} supports other settings as
+ * well, which this configuration object could help set.
  */
 @InternalExtensionOnly
 public class CallOptionsConfig implements Serializable {
@@ -53,9 +57,12 @@ public class CallOptionsConfig implements Serializable {
   public static class Builder {
     private boolean useTimeout = USE_TIMEOUT_DEFAULT;
     private int shortRpcTimeoutMs = SHORT_TIMEOUT_MS_DEFAULT;
+    private Optional<Integer> shortRpcAttemptTimeoutMs = Optional.absent();
     private int longRpcTimeoutMs = LONG_TIMEOUT_MS_DEFAULT;
     private int mutateRpcTimeoutMs = LONG_TIMEOUT_MS_DEFAULT;
+    private Optional<Integer> mutateRpcAttemptTimeoutMs = Optional.absent();
     private int readRowsRpcTimeoutMs = LONG_TIMEOUT_MS_DEFAULT;
+    private Optional<Integer> readRowsRpcAttemptTimeoutMs = Optional.absent();
 
     /** @deprecated Please use {@link CallOptionsConfig#builder()} */
     @Deprecated
@@ -64,9 +71,12 @@ public class CallOptionsConfig implements Serializable {
     private Builder(CallOptionsConfig original) {
       this.useTimeout = original.useTimeout;
       this.shortRpcTimeoutMs = original.shortRpcTimeoutMs;
+      this.shortRpcAttemptTimeoutMs = original.shortRpcAttemptTimeoutMs;
       this.longRpcTimeoutMs = original.longRpcTimeoutMs;
       this.mutateRpcTimeoutMs = original.mutateRpcTimeoutMs;
+      this.mutateRpcAttemptTimeoutMs = original.mutateRpcAttemptTimeoutMs;
       this.readRowsRpcTimeoutMs = original.readStreamRpcTimeoutMs;
+      this.readRowsRpcAttemptTimeoutMs = original.readStreamRpcAttemptTimeoutMs;
     }
 
     /**
@@ -85,6 +95,7 @@ public class CallOptionsConfig implements Serializable {
      *
      * @param timeoutMs timeout value in milliseconds.
      * @return a {@link Builder} object, for chaining
+     * @deprecated Use {@link #setShortRpcAttemptTimeoutMs(int)} instead
      */
     @Deprecated
     public Builder setTimeoutMs(int timeoutMs) {
@@ -92,7 +103,14 @@ public class CallOptionsConfig implements Serializable {
     }
 
     /**
-     * The amount of milliseconds to wait before issuing a client side timeout for short RPCs.
+     * The amount of milliseconds to wait before issuing a client side timeout for a short RPC
+     * operation.
+     *
+     * <p>By short RPC, we mean single-row operations like CheckAndMutateRow, MutateRow, and
+     * ReadRows on a single key.
+     *
+     * <p>The timeout spans the full end-to-end operation from the user perspective, i.e. covering
+     * multiple retry attempts.
      *
      * @param shortRpcTimeoutMs timeout value in milliseconds.
      * @return a {@link Builder} object, for chaining
@@ -104,6 +122,26 @@ public class CallOptionsConfig implements Serializable {
       return this;
     }
 
+    /**
+     * The amount of milliseconds to wait before issuing a client side timeout for a short RPC
+     * attempt.
+     *
+     * <p>See {@link #setShortRpcTimeoutMs(int)} for the definition of short RPC.
+     *
+     * <p>This timeout covers a single RPC call attempt to the server, whereas {@link
+     * #setShortRpcTimeoutMs(int)} covers the total time across all attempts.
+     *
+     * @param shortRpcAttemptTimeoutMs timeout value in milliseconds.
+     * @return a {@link Builder} object, for chaining
+     */
+    @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+    public Builder setShortRpcAttemptTimeoutMs(int shortRpcAttemptTimeoutMs) {
+      Preconditions.checkArgument(
+          shortRpcAttemptTimeoutMs > 0, "Short Timeout ms has to be greater than 0.");
+      this.shortRpcAttemptTimeoutMs = Optional.of(shortRpcAttemptTimeoutMs);
+      return this;
+    }
+
     /**
      * The amount of milliseconds to wait before issuing a client side timeout for long running
      * RPCs.
@@ -122,8 +160,12 @@ public class CallOptionsConfig implements Serializable {
     }
 
     /**
-     * The amount of time in milliseconds to wait before issuing a client side timeout for row
-     * mutation RPCs.
+     * The amount of time in milliseconds to wait before issuing a client side timeout for bulk row
+     * mutation RPCs (i.e. multiple rows). Single-row mutations are governed by {@link
+     * #shortRpcTimeoutMs};
+     *
+     * <p>The timeout spans the full end-to-end operation from the user perspective, i.e. covering
+     * multiple retry attempts.
      *
      * @param mutateRpcTimeoutMs timeout value in milliseconds.
      * @return a {@link Builder} object, for chaining
@@ -135,10 +177,32 @@ public class CallOptionsConfig implements Serializable {
       return this;
     }
 
+    /**
+     * The amount of time in milliseconds to wait before issuing a client side timeout for bulk row
+     * mutation RPCs (i.e. multiple rows). Single-row mutations are governed by {@link
+     * #shortRpcAttemptTimeoutMs};
+     *
+     * <p>This timeout covers a single RPC call attempt to the server, whereas {@link
+     * #setShortRpcTimeoutMs(int)} covers the total time across all attempts.
+     *
+     * @param mutateRpcAttemptTimeoutMs timeout value in milliseconds.
+     * @return a {@link Builder} object, for chaining
+     */
+    @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+    public Builder setMutateRpcAttemptTimeoutMs(int mutateRpcAttemptTimeoutMs) {
+      Preconditions.checkArgument(
+          mutateRpcAttemptTimeoutMs > 0, "Mutate Rows RPC Timeout ms has to be greater than 0");
+      this.mutateRpcAttemptTimeoutMs = Optional.of(mutateRpcAttemptTimeoutMs);
+      return this;
+    }
+
     /**
      * The amount of time in millisecond to wait before issuing a client side timeout for readRows
      * streaming RPCs.
      *
+     * <p>The timeout spans the full end-to-end operation from the user perspective, i.e. covering
+     * multiple retry attempts.
+     *
      * @param readStreamRpcTimeoutMs timeout value in milliseconds.
      * @return a {@link Builder} object, for chaining
      */
@@ -149,6 +213,24 @@ public class CallOptionsConfig implements Serializable {
       return this;
     }
 
+    /**
+     * The amount of time in milliseconds to wait before issuing a client side timeout for readRows
+     * streaming RPCs.
+     *
+     * <p>This timeout covers a single RPC call attempt to the server, whereas {@link
+     * #setShortRpcTimeoutMs(int)} covers the total time across all attempts.
+     *
+     * @param readRowsRpcAttemptTimeoutMs timeout value in milliseconds.
+     * @return a {@link Builder} object, for chaining
+     */
+    @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+    public Builder setReadRowsRpcAttemptTimeoutMs(int readRowsRpcAttemptTimeoutMs) {
+      Preconditions.checkArgument(
+          readRowsRpcAttemptTimeoutMs > 0, "Mutate Rows RPC Timeout ms has to be greater than 0");
+      this.readRowsRpcAttemptTimeoutMs = Optional.of(readRowsRpcAttemptTimeoutMs);
+      return this;
+    }
+
     public CallOptionsConfig build() {
       return new CallOptionsConfig(this);
     }
@@ -156,9 +238,12 @@ public class CallOptionsConfig implements Serializable {
 
   private final boolean useTimeout;
   private final int shortRpcTimeoutMs;
+  private final Optional<Integer> shortRpcAttemptTimeoutMs;
   private final int longRpcTimeoutMs;
   private final int mutateRpcTimeoutMs;
+  private final Optional<Integer> mutateRpcAttemptTimeoutMs;
   private final int readStreamRpcTimeoutMs;
+  private final Optional<Integer> readStreamRpcAttemptTimeoutMs;
 
   /**
    * Constructor for CallOptionsConfig.
@@ -172,14 +257,18 @@ public class CallOptionsConfig implements Serializable {
   public CallOptionsConfig(boolean useTimeout, int unaryRpcTimeoutMs, int longRpcTimeoutMs) {
     this.useTimeout = useTimeout;
     this.shortRpcTimeoutMs = unaryRpcTimeoutMs;
+    this.shortRpcAttemptTimeoutMs = Optional.absent();
     this.longRpcTimeoutMs = longRpcTimeoutMs;
     this.mutateRpcTimeoutMs = longRpcTimeoutMs;
+    this.mutateRpcAttemptTimeoutMs = Optional.absent();
     this.readStreamRpcTimeoutMs = longRpcTimeoutMs;
+    this.readStreamRpcAttemptTimeoutMs = Optional.absent();
   }
 
   private CallOptionsConfig(Builder builder) {
     this.useTimeout = builder.useTimeout;
     this.shortRpcTimeoutMs = builder.shortRpcTimeoutMs;
+    this.shortRpcAttemptTimeoutMs = builder.shortRpcAttemptTimeoutMs;
     this.longRpcTimeoutMs = builder.longRpcTimeoutMs;
     int mutateTimeout = builder.mutateRpcTimeoutMs;
     int readRowsTimeout = builder.readRowsRpcTimeoutMs;
@@ -191,7 +280,9 @@ public class CallOptionsConfig implements Serializable {
       readRowsTimeout = longRpcTimeoutMs;
     }
     this.mutateRpcTimeoutMs = mutateTimeout;
+    this.mutateRpcAttemptTimeoutMs = builder.mutateRpcAttemptTimeoutMs;
     this.readStreamRpcTimeoutMs = readRowsTimeout;
+    this.readStreamRpcAttemptTimeoutMs = builder.readRowsRpcAttemptTimeoutMs;
   }
 
   /**
@@ -223,6 +314,19 @@ public class CallOptionsConfig implements Serializable {
     return shortRpcTimeoutMs;
   }
 
+  /**
+   * Getter for the field <code>shortRpcAttemptTimeoutMs</code>.
+   *
+   * <p>If the Optional value is not present, the attempt timeout will leverage {@link
+   * #shortRpcTimeoutMs}.
+   *
+   * @return an Optional int, populated if the property was present.
+   */
+  @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+  public Optional<Integer> getShortRpcAttemptTimeoutMs() {
+    return shortRpcAttemptTimeoutMs;
+  }
+
   /**
    * Getter for the field <code>longRpcTimeoutMs</code>.
    *
@@ -244,6 +348,19 @@ public class CallOptionsConfig implements Serializable {
     return mutateRpcTimeoutMs;
   }
 
+  /**
+   * Getter for the field <code>mutateRpcAttemptTimeoutMs</code>.
+   *
+   * <p>If the Optional value is not present, the attempt timeout will leverage {@link
+   * #mutateRpcTimeoutMs}.
+   *
+   * @return an Optional int, populated if the property was present.
+   */
+  @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+  public Optional<Integer> getMutateRpcAttemptTimeoutMs() {
+    return mutateRpcAttemptTimeoutMs;
+  }
+
   /**
    * Getter for the field <code>readStreamRpcTimeoutMs</code>.
    *
@@ -253,6 +370,19 @@ public class CallOptionsConfig implements Serializable {
     return readStreamRpcTimeoutMs;
   }
 
+  /**
+   * Getter for the field <code>readStreamRpcAttemptTimeoutMs</code>.
+   *
+   * <p>If the Optional value is not present, the attempt timeout will leverage {@link
+   * #readStreamRpcTimeoutMs}.
+   *
+   * @return an Optional int, populated if the property was present.
+   */
+  @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+  public Optional<Integer> getReadStreamRpcAttemptTimeoutMs() {
+    return readStreamRpcAttemptTimeoutMs;
+  }
+
   /** {@inheritDoc} */
   @Override
   public boolean equals(Object obj) {
@@ -265,9 +395,25 @@ public class CallOptionsConfig implements Serializable {
     CallOptionsConfig other = (CallOptionsConfig) obj;
     return useTimeout == other.useTimeout
         && shortRpcTimeoutMs == other.shortRpcTimeoutMs
+        && Objects.equals(shortRpcAttemptTimeoutMs, other.shortRpcAttemptTimeoutMs)
         && longRpcTimeoutMs == other.longRpcTimeoutMs
         && mutateRpcTimeoutMs == other.mutateRpcTimeoutMs
-        && readStreamRpcTimeoutMs == other.readStreamRpcTimeoutMs;
+        && Objects.equals(mutateRpcAttemptTimeoutMs, other.mutateRpcAttemptTimeoutMs)
+        && readStreamRpcTimeoutMs == other.readStreamRpcTimeoutMs
+        && Objects.equals(readStreamRpcAttemptTimeoutMs, other.readStreamRpcAttemptTimeoutMs);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        useTimeout,
+        shortRpcTimeoutMs,
+        shortRpcAttemptTimeoutMs,
+        longRpcTimeoutMs,
+        mutateRpcTimeoutMs,
+        mutateRpcAttemptTimeoutMs,
+        readStreamRpcTimeoutMs,
+        readStreamRpcAttemptTimeoutMs);
   }
 
   /** {@inheritDoc} */
@@ -276,9 +422,12 @@ public class CallOptionsConfig implements Serializable {
     return MoreObjects.toStringHelper(this)
         .add("useTimeout", useTimeout)
         .add("shortRpcTimeoutMs", shortRpcTimeoutMs)
+        .add("shortRpcAttemptTimeoutMs", shortRpcAttemptTimeoutMs)
         .add("longRpcTimeoutMs", longRpcTimeoutMs)
         .add("mutateRpcTimeoutMs", mutateRpcTimeoutMs)
+        .add("mutateRpcAttemptTimeoutMs", mutateRpcAttemptTimeoutMs)
         .add("readStreamRpcTimeoutMs", readStreamRpcTimeoutMs)
+        .add("readStreamRpcAttemptTimeoutMs", readStreamRpcAttemptTimeoutMs)
         .toString();
   }
 
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClient.java
index 7e786069d..0acf3810b 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClient.java
@@ -183,11 +183,10 @@ public interface BigtableDataClient {
   ListenableFuture<List<FlatRow>> readFlatRowsAsync(ReadRowsRequest request);
 
   /**
-   * Sets a {@link com.google.cloud.bigtable.grpc.CallOptionsFactory} which creates {@link
-   * io.grpc.CallOptions}
+   * Sets a {@link DeadlineGeneratorFactory} which creates {@link io.grpc.CallOptions}
    *
    * <p>For internal use only - public for technical reasons.
    */
   @InternalApi("For internal usage only")
-  void setCallOptionsFactory(CallOptionsFactory callOptionsFactory);
+  void setDeadlineGeneratorFactory(DeadlineGeneratorFactory deadlineGeneratorFactory);
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java
index 9cf73ae73..f807c260e 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataGrpcClient.java
@@ -59,10 +59,8 @@ import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
-import io.grpc.CallOptions;
 import io.grpc.Channel;
 import io.grpc.Metadata;
-import io.grpc.MethodDescriptor;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
 import java.util.List;
@@ -153,7 +151,7 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   private final ScheduledExecutorService retryExecutorService;
   private final RetryOptions retryOptions;
 
-  private CallOptionsFactory callOptionsFactory = new CallOptionsFactory.Default();
+  private DeadlineGeneratorFactory deadlineGeneratorFactory = DeadlineGeneratorFactory.DEFAULT;
 
   private final BigtableAsyncRpc<SampleRowKeysRequest, SampleRowKeysResponse> sampleRowKeysAsync;
   private final BigtableAsyncRpc<ReadRowsRequest, ReadRowsResponse> readRowsAsync;
@@ -217,8 +215,8 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
 
   /** {@inheritDoc} */
   @Override
-  public void setCallOptionsFactory(CallOptionsFactory callOptionsFactory) {
-    this.callOptionsFactory = callOptionsFactory;
+  public void setDeadlineGeneratorFactory(DeadlineGeneratorFactory deadlineGeneratorFactory) {
+    this.deadlineGeneratorFactory = deadlineGeneratorFactory;
   }
 
   private <T> Predicate<T> getMutationRetryableFunction(Predicate<T> isRetryableMutation) {
@@ -274,10 +272,17 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
     if (shouldOverrideAppProfile(request.getAppProfileId())) {
       request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
     }
-    CallOptions callOptions = getCallOptions(mutateRowsRpc.getMethodDescriptor(), request);
+    DeadlineGenerator deadlineGenerator =
+        deadlineGeneratorFactory.getRequestDeadlineGenerator(request);
     Metadata metadata = createMetadata(request.getTableName());
     return new RetryingMutateRowsOperation(
-        retryOptions, request, mutateRowsRpc, callOptions, retryExecutorService, metadata, CLOCK);
+        retryOptions,
+        request,
+        mutateRowsRpc,
+        deadlineGenerator,
+        retryExecutorService,
+        metadata,
+        CLOCK);
   }
 
   /** {@inheritDoc} */
@@ -388,23 +393,20 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
 
   private <ReqT, RespT> RetryingUnaryOperation<ReqT, RespT> createUnaryListener(
       ReqT request, BigtableAsyncRpc<ReqT, RespT> rpc, String tableName) {
-    CallOptions callOptions = getCallOptions(rpc.getMethodDescriptor(), request);
+    DeadlineGenerator deadlineGenerator =
+        deadlineGeneratorFactory.getRequestDeadlineGenerator(request);
     Metadata metadata = createMetadata(tableName);
     return new RetryingUnaryOperation<>(
-        retryOptions, request, rpc, callOptions, retryExecutorService, metadata, CLOCK);
+        retryOptions, request, rpc, deadlineGenerator, retryExecutorService, metadata, CLOCK);
   }
 
   private <ReqT, RespT> RetryingStreamOperation<ReqT, RespT> createStreamingListener(
       ReqT request, BigtableAsyncRpc<ReqT, RespT> rpc, String tableName) {
-    CallOptions callOptions = getCallOptions(rpc.getMethodDescriptor(), request);
+    DeadlineGenerator deadlineGenerator =
+        deadlineGeneratorFactory.getRequestDeadlineGenerator(request);
     Metadata metadata = createMetadata(tableName);
     return new RetryingStreamOperation<>(
-        retryOptions, request, rpc, callOptions, retryExecutorService, metadata, CLOCK);
-  }
-
-  private <ReqT> CallOptions getCallOptions(
-      final MethodDescriptor<ReqT, ?> methodDescriptor, ReqT request) {
-    return callOptionsFactory.create(methodDescriptor, request);
+        retryOptions, request, rpc, deadlineGenerator, retryExecutorService, metadata, CLOCK);
   }
 
   /** Creates a {@link Metadata} that contains pertinent headers. */
@@ -512,7 +514,7 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
         retryOptions,
         request,
         readRowsAsync,
-        getCallOptions(readRowsAsync.getMethodDescriptor(), request),
+        deadlineGeneratorFactory.getRequestDeadlineGenerator(request),
         retryExecutorService,
         createMetadata(request.getTableName()),
         CLOCK);
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
index c95d19d36..057293540 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
@@ -343,13 +343,13 @@ public class BigtableSession implements Closeable {
 
       BigtableSessionSharedThreadPools sharedPools = BigtableSessionSharedThreadPools.getInstance();
 
-      CallOptionsFactory.ConfiguredCallOptionsFactory callOptionsFactory =
-          new CallOptionsFactory.ConfiguredCallOptionsFactory(options.getCallOptionsConfig());
+      ConfiguredDeadlineGeneratorFactory callOptionsFactory =
+          new ConfiguredDeadlineGeneratorFactory(options.getCallOptionsConfig());
 
       // More often than not, users want the dataClient. Create a new one in the constructor.
       this.dataClient =
           new BigtableDataGrpcClient(dataChannel, sharedPools.getRetryExecutor(), options);
-      this.dataClient.setCallOptionsFactory(callOptionsFactory);
+      this.dataClient.setDeadlineGeneratorFactory(callOptionsFactory);
 
       // Async operations can run amok, so they need to have some throttling. The throttling is
       // achieved through a ThrottlingClientInterceptor.  gRPC wraps ClientInterceptors in Channels,
@@ -364,7 +364,7 @@ public class BigtableSession implements Closeable {
               dataChannel, new ThrottlingClientInterceptor(resourceLimiter));
       throttlingDataClient =
           new BigtableDataGrpcClient(asyncDataChannel, sharedPools.getRetryExecutor(), options);
-      throttlingDataClient.setCallOptionsFactory(callOptionsFactory);
+      throttlingDataClient.setDeadlineGeneratorFactory(callOptionsFactory);
 
       ManagedChannel rawAdminChannel = createNettyChannel(options.getAdminHost(), options);
       managedChannels.add(rawAdminChannel);
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java
index 12e726c27..302668692 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java
@@ -68,7 +68,6 @@ import com.google.longrunning.GetOperationRequest;
 import com.google.longrunning.Operation;
 import com.google.longrunning.OperationsGrpc;
 import com.google.protobuf.Empty;
-import io.grpc.CallOptions;
 import io.grpc.Channel;
 import io.grpc.Metadata;
 import java.io.IOException;
@@ -83,6 +82,8 @@ import java.util.concurrent.TimeoutException;
  */
 @InternalApi("For internal usage only")
 public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
+  private final DeadlineGeneratorFactory deadlineGeneratorFactory =
+      DeadlineGeneratorFactory.DEFAULT;
 
   private final OperationUtil operationUtil;
 
@@ -447,13 +448,12 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
 
   private <ReqT, RespT> RetryingUnaryOperation<ReqT, RespT> createUnaryListener(
       ReqT request, BigtableAsyncRpc<ReqT, RespT> rpc, String resource) {
-    CallOptions callOptions = CallOptions.DEFAULT;
     Metadata metadata = createMetadata(resource);
     return new RetryingUnaryOperation<>(
         retryOptions,
         request,
         rpc,
-        callOptions,
+        deadlineGeneratorFactory.getRequestDeadlineGenerator(request),
         retryExecutorService,
         metadata,
         NanoClock.getDefaultClock());
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/CallOptionsFactory.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/CallOptionsFactory.java
deleted file mode 100644
index d73e75b52..000000000
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/CallOptionsFactory.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * Copyright 2015 Google Inc. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file 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 com.google.cloud.bigtable.grpc;
-
-import com.google.api.core.InternalApi;
-import com.google.bigtable.v2.MutateRowsRequest;
-import com.google.bigtable.v2.ReadRowsRequest;
-import com.google.bigtable.v2.RowSet;
-import com.google.cloud.bigtable.config.CallOptionsConfig;
-import io.grpc.CallOptions;
-import io.grpc.Context;
-import io.grpc.Deadline;
-import io.grpc.MethodDescriptor;
-import java.util.concurrent.TimeUnit;
-
-/**
- * A factory that creates {@link CallOptions} for use in {@link BigtableDataClient} RPCs.
- *
- * <p>For internal use only - public for technical reasons.
- */
-@InternalApi("For internal usage only")
-public interface CallOptionsFactory {
-
-  /**
-   * Provide a {@link CallOptions} object to be used in a single RPC. {@link CallOptions} can
-   * contain state, specifically start time with an expiration is set; in cases when timeouts are
-   * used, implementations should create a new CallOptions each time this method is called.
-   */
-  <RequestT> CallOptions create(MethodDescriptor<RequestT, ?> descriptor, RequestT request);
-
-  /**
-   * Returns {@link CallOptions#DEFAULT} with any {@link Context#current()}'s {@link
-   * Context#getDeadline()} applied to it.
-   *
-   * <p>For internal use only - public for technical reasons.
-   */
-  @InternalApi("For internal usage only")
-  public static class Default implements CallOptionsFactory {
-    @Override
-    public <RequestT> CallOptions create(
-        MethodDescriptor<RequestT, ?> descriptor, RequestT request) {
-      Deadline contextDeadline = Context.current().getDeadline();
-      if (contextDeadline != null) {
-        return CallOptions.DEFAULT.withDeadline(contextDeadline);
-      } else {
-        return CallOptions.DEFAULT;
-      }
-    }
-  }
-
-  /**
-   * Creates a new {@link CallOptions} based on a {@link CallOptionsConfig}.
-   *
-   * <p>For internal use only - public for technical reasons.
-   */
-  @InternalApi("For internal usage only")
-  public static class ConfiguredCallOptionsFactory implements CallOptionsFactory {
-    private final CallOptionsConfig config;
-
-    public ConfiguredCallOptionsFactory(CallOptionsConfig config) {
-      this.config = config;
-    }
-
-    /**
-     * Creates a {@link CallOptions} with a focus on {@link Deadline}. Deadlines are decided in the
-     * following order:
-     *
-     * <ol>
-     *   <li>If a user set a {@link Context} deadline (see {@link Context#getDeadline()}), use that
-     *   <li>If a user configured deadlines via {@link CallOptionsConfig}, use it.
-     *   <li>Otherwise, use {@link CallOptions#DEFAULT}.
-     * </ol>
-     */
-    @Override
-    public <RequestT> CallOptions create(
-        MethodDescriptor<RequestT, ?> descriptor, RequestT request) {
-      Deadline contextDeadline = Context.current().getDeadline();
-      if (contextDeadline != null) {
-        return CallOptions.DEFAULT.withDeadline(contextDeadline);
-      } else if (config.isUseTimeout() && request != null) {
-        int timeout = getRequestTimeout(request);
-        return CallOptions.DEFAULT.withDeadline(Deadline.after(timeout, TimeUnit.MILLISECONDS));
-      } else {
-        return CallOptions.DEFAULT;
-      }
-    }
-
-    /**
-     * @param request an RPC request.
-     * @return timeout in milliseconds as per the Request type.
-     */
-    private int getRequestTimeout(Object request) {
-      if (request instanceof ReadRowsRequest && !isGet((ReadRowsRequest) request)) {
-        return config.getReadStreamRpcTimeoutMs();
-      } else if (request instanceof MutateRowsRequest) {
-        return config.getMutateRpcTimeoutMs();
-      } else {
-        return config.getShortRpcTimeoutMs();
-      }
-    }
-
-    /**
-     * @param request
-     * @return true if this is a {@link MutateRowsRequest} or a {@link ReadRowsRequest} that's a
-     *     scan.
-     * @deprecated Please use {@link #getRequestTimeout(Object)} to fetch long requests timeout.
-     */
-    @Deprecated
-    public static boolean isLongRequest(Object request) {
-      if (request instanceof ReadRowsRequest) {
-        return !isGet((ReadRowsRequest) request);
-      } else {
-        return request instanceof MutateRowsRequest;
-      }
-    }
-
-    public static boolean isGet(ReadRowsRequest request) {
-      RowSet rowSet = request.getRows();
-      return rowSet != null && rowSet.getRowRangesCount() == 0 && rowSet.getRowKeysCount() == 1;
-    }
-  }
-}
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/ConfiguredDeadlineGeneratorFactory.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/ConfiguredDeadlineGeneratorFactory.java
new file mode 100644
index 000000000..f7b0294c8
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/ConfiguredDeadlineGeneratorFactory.java
@@ -0,0 +1,163 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.grpc;
+
+import static com.google.cloud.bigtable.grpc.DeadlineGenerator.UNARY_DEADLINE_MINUTES;
+
+import com.google.api.core.InternalApi;
+import com.google.bigtable.v2.MutateRowsRequest;
+import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.bigtable.v2.RowSet;
+import com.google.cloud.bigtable.config.CallOptionsConfig;
+import com.google.common.base.Optional;
+import io.grpc.CallOptions;
+import io.grpc.Context;
+import io.grpc.Deadline;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A {@link DeadlineGeneratorFactory} instance based on the configurations from {@link
+ * CallOptionsConfig}.
+ *
+ * <p>For internal use only - public for technical reasons.
+ */
+@InternalApi("For internal usage only")
+public class ConfiguredDeadlineGeneratorFactory implements DeadlineGeneratorFactory {
+  private final CallOptionsConfig config;
+
+  public ConfiguredDeadlineGeneratorFactory(CallOptionsConfig config) {
+    this.config = config;
+  }
+
+  @Override
+  public <RequestT> DeadlineGenerator getRequestDeadlineGenerator(RequestT request) {
+    if (isStreamingRead(request)) {
+      return new DeadlineGeneratorImpl(
+          config.isUseTimeout(),
+          config.getReadStreamRpcTimeoutMs(),
+          config.getReadStreamRpcAttemptTimeoutMs(),
+          // Streaming reads do not need to specify a fallback deadline, since the Watchdog will
+          // take affect and ensure that hanging does not occur. See
+          // com.google.cloud.bigtable.grpc.io.Watchdog, which handles hanging for streaming reads.
+          Optional.<Long>absent());
+    } else if (request instanceof MutateRowsRequest) {
+      return new DeadlineGeneratorImpl(
+          config.isUseTimeout(),
+          config.getMutateRpcTimeoutMs(),
+          config.getMutateRpcAttemptTimeoutMs(),
+          // Calls should fail after 6 minutes, if there isn't any response from the server.
+          Optional.of(TimeUnit.MINUTES.toMillis(UNARY_DEADLINE_MINUTES)));
+    } else {
+      return new DeadlineGeneratorImpl(
+          config.isUseTimeout(),
+          config.getShortRpcTimeoutMs(),
+          config.getShortRpcAttemptTimeoutMs(),
+          // Calls should fail after 6 minutes, if there isn't any response from the server.
+          Optional.of(TimeUnit.MINUTES.toMillis(UNARY_DEADLINE_MINUTES)));
+    }
+  }
+
+  private static <RequestT> boolean isStreamingRead(RequestT request) {
+    return request instanceof ReadRowsRequest && !isGet((ReadRowsRequest) request);
+  }
+
+  private static boolean isGet(ReadRowsRequest request) {
+    RowSet rowSet = request.getRows();
+    return rowSet != null && rowSet.getRowRangesCount() == 0 && rowSet.getRowKeysCount() == 1;
+  }
+
+  /**
+   * Provide a {@link CallOptions} object to be used in a single RPC attempt, where an RPC operation
+   * may have multiple RPC attempts. {@link CallOptions} can contain state, specifically start time
+   * with an expiration is set; in cases when timeouts are used, implementations should create a new
+   * CallOptions each time this method is called.
+   */
+  private static class DeadlineGeneratorImpl implements DeadlineGenerator {
+    private final boolean useTimeout;
+    private final int requestTimeoutMs;
+    private final Optional<Integer> requestAttemptTimeoutMs;
+    /** A fail safe RPC deadline to make sure that operations don't hang. */
+    private final Optional<Long> nonTimeoutFallbackDeadlineMs;
+
+    private final Optional<Deadline> operationDeadline;
+
+    DeadlineGeneratorImpl(
+        boolean useTimeout,
+        int requestTimeoutMs,
+        Optional<Integer> requestAttemptTimeoutMs,
+        Optional<Long> nonTimeoutFallbackDeadlineMs) {
+      this.useTimeout = useTimeout;
+      this.requestTimeoutMs = requestTimeoutMs;
+      this.requestAttemptTimeoutMs = requestAttemptTimeoutMs;
+      this.nonTimeoutFallbackDeadlineMs = nonTimeoutFallbackDeadlineMs;
+      // Note: operationDeadline is always non-null if useTimeout == true
+      this.operationDeadline = createOperationDeadline();
+    }
+
+    /**
+     * Creates a {@link Deadline} for the operation, decided in the following order:
+     *
+     * <ol>
+     *   <li>If a user set a {@link Context} deadline (see {@link Deadline}), use that
+     *   <li>If a user configured deadlines via {@link CallOptionsConfig}, use it.
+     *   <li>Otherwise, no deadline is specified.
+     * </ol>
+     */
+    private Optional<Deadline> createOperationDeadline() {
+      Optional<Deadline> deadline = Optional.fromNullable(Context.current().getDeadline());
+      if (useTimeout) {
+        deadline =
+            deadline.or(Optional.of(Deadline.after(requestTimeoutMs, TimeUnit.MILLISECONDS)));
+      }
+      return deadline;
+    }
+
+    @Override
+    public Optional<Long> getOperationTimeoutMs() {
+      return operationDeadline.isPresent()
+          ? Optional.of(operationDeadline.get().timeRemaining(TimeUnit.MILLISECONDS))
+          : Optional.<Long>absent();
+    }
+
+    @Override
+    public Optional<Deadline> getRpcAttemptDeadline() {
+      // Check for RPC Attempt Timeouts first for backward-compatibility, so that
+      // the existing logic (based mostly on the operation timeout) is preserved.
+      if (useTimeout && requestAttemptTimeoutMs.isPresent()) {
+        Deadline attemptDeadline =
+            Deadline.after(requestAttemptTimeoutMs.get(), TimeUnit.MILLISECONDS);
+        // Ensure we don't exceed the absolute operation deadline by taking the min of the attempt
+        // and operation deadlines.
+        if (operationDeadline.isPresent()) {
+          return Optional.of(attemptDeadline.minimum(operationDeadline.get()));
+        } else {
+          // This branch should never be exercised, since if useTimeout == true, then we've set the
+          // operationDeadline in the constructor. To avoid any compile-time syntax questions and to
+          // add documentation, we keep this code.
+          return Optional.of(attemptDeadline);
+        }
+      } else if (operationDeadline.isPresent()) {
+        // If the user set a deadline, honor it.
+        return operationDeadline;
+      } else if (nonTimeoutFallbackDeadlineMs.isPresent()) {
+        return Optional.of(
+            Deadline.after(nonTimeoutFallbackDeadlineMs.get(), TimeUnit.MILLISECONDS));
+      } else {
+        return Optional.absent();
+      }
+    }
+  }
+}
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/DeadlineGenerator.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/DeadlineGenerator.java
new file mode 100644
index 000000000..020cd6c18
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/DeadlineGenerator.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.grpc;
+
+import com.google.api.core.InternalApi;
+import com.google.common.base.Optional;
+import io.grpc.CallOptions;
+import io.grpc.Context;
+import io.grpc.Deadline;
+
+/**
+ * A provider for RPC timeouts and deadlines for a given Bigtable client operation, governing both
+ * the full operation and retry attempts within an operation. A single {@link DeadlineGenerator}
+ * object should be used for a client operation instance and reused across retry attempts.
+ */
+@InternalApi("For internal usage only")
+public interface DeadlineGenerator {
+  /**
+   * The server-side has a 5 minute timeout. Unary operations should be timed-out on the client side
+   * after 6 minutes.
+   */
+  long UNARY_DEADLINE_MINUTES = 6L;
+
+  /**
+   * The default instance will return: - {@link CallOptions#DEFAULT} with any {@link
+   * Context#current()}'s {@link Context#getDeadline()} applied to it. - The operation timeout of 6
+   * minutes.
+   *
+   * <p>For internal use only - public for technical reasons.
+   */
+  DeadlineGenerator DEFAULT =
+      new DeadlineGenerator() {
+        @Override
+        public Optional<Long> getOperationTimeoutMs() {
+          return Optional.absent();
+        }
+
+        @Override
+        public Optional<Deadline> getRpcAttemptDeadline() {
+          return Optional.fromNullable(Context.current().getDeadline());
+        }
+      };
+
+  /**
+   * Returns the operation timeout (if configured) in millisecond to use for this RPC, covering all
+   * retry attempts. Single attempt timeouts are governed by {@link #getRpcAttemptDeadline()} ()}.
+   */
+  Optional<Long> getOperationTimeoutMs();
+
+  /**
+   * Provides a {@link Deadline} object to be used in a single RPC attempt. {@link Deadline} can
+   * contain state, specifically start time with an expiration is set; in cases when timeouts are
+   * used, implementations should invoke this method each time this method is called.
+   */
+  Optional<Deadline> getRpcAttemptDeadline();
+}
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/DeadlineGeneratorFactory.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/DeadlineGeneratorFactory.java
new file mode 100644
index 000000000..1805ee493
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/DeadlineGeneratorFactory.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2015 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.grpc;
+
+import com.google.api.core.InternalApi;
+
+/**
+ * A factory that creates {@link DeadlineGenerator} for use in {@link BigtableDataClient} RPCs.
+ *
+ * <p>For internal use only - public for technical reasons.
+ */
+@InternalApi("For internal usage only")
+public interface DeadlineGeneratorFactory {
+  DeadlineGeneratorFactory DEFAULT =
+      new DeadlineGeneratorFactory() {
+        @Override
+        public <RequestT> DeadlineGenerator getRequestDeadlineGenerator(RequestT request) {
+          return DeadlineGenerator.DEFAULT;
+        }
+      };
+
+  /**
+   * Returns a {@link DeadlineGenerator} instance to use for the given RPC operation. This should
+   * only be called once per operation and be used for the duration of the operation.
+   */
+  <RequestT> DeadlineGenerator getRequestDeadlineGenerator(RequestT request);
+}
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
index fbedb2fdc..9f943d4f5 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
@@ -20,14 +20,14 @@ import com.google.api.core.InternalApi;
 import com.google.api.gax.retrying.ExponentialRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.retrying.TimedAttemptSettings;
-import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.cloud.bigtable.config.Logger;
 import com.google.cloud.bigtable.config.RetryOptions;
-import com.google.cloud.bigtable.grpc.CallOptionsFactory;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
 import com.google.cloud.bigtable.grpc.io.ChannelPool;
 import com.google.cloud.bigtable.grpc.scanner.BigtableRetriesExhaustedException;
 import com.google.cloud.bigtable.metrics.Timer;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.AbstractFuture;
@@ -71,10 +71,6 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
   private static final EndSpanOptions END_SPAN_OPTIONS_WITH_SAMPLE_STORE =
       EndSpanOptions.builder().setSampleToLocalSpanStore(true).build();
 
-  // The server-side has a 5 minute timeout. Unary operations should be timed-out on the client side
-  // after 6 minutes.
-  protected static final long UNARY_DEADLINE_MINUTES = 6l;
-
   private static String makeSpanName(String prefix, String fullMethodName) {
     return prefix + "." + fullMethodName.replace('/', '.');
   }
@@ -111,7 +107,12 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
   protected final ScheduledExecutorService retryExecutorService;
 
   private final RequestT request;
-  private final CallOptions callOptions;
+  /**
+   * gRPC callOptionsFactory meant to use for both the full RPC operation and the individual RPC
+   * attempts.
+   */
+  private final DeadlineGenerator deadlineGenerator;
+
   private final Metadata originalMetadata;
 
   protected int failedCount = 0;
@@ -131,7 +132,7 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
    * @param retryOptions a {@link com.google.cloud.bigtable.config.RetryOptions} object.
    * @param request a RequestT object.
    * @param retryableRpc a {@link com.google.cloud.bigtable.grpc.async.BigtableAsyncRpc} object.
-   * @param callOptions a {@link io.grpc.CallOptions} object.
+   * @param deadlineGenerator a {@link DeadlineGenerator} object.
    * @param retryExecutorService a {@link java.util.concurrent.ScheduledExecutorService} object.
    * @param originalMetadata a {@link io.grpc.Metadata} object.
    * @param clock a {@link ApiClock} object
@@ -140,14 +141,14 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
       RetryOptions retryOptions,
       RequestT request,
       BigtableAsyncRpc<RequestT, ResponseT> retryableRpc,
-      CallOptions callOptions,
+      DeadlineGenerator deadlineGenerator,
       ScheduledExecutorService retryExecutorService,
       Metadata originalMetadata,
       ApiClock clock) {
     this.retryOptions = retryOptions;
     this.request = request;
     this.rpc = retryableRpc;
-    this.callOptions = callOptions;
+    this.deadlineGenerator = deadlineGenerator;
     this.retryExecutorService = retryExecutorService;
     this.originalMetadata = originalMetadata;
     this.completionFuture = new GrpcFuture<>();
@@ -323,12 +324,11 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
    * @return a {@link ExponentialRetryAlgorithm} object.
    */
   private ExponentialRetryAlgorithm createRetryAlgorithm(ApiClock clock) {
-    long timeoutMs = retryOptions.getMaxElapsedBackoffMillis();
+    Optional<Long> operationTimeoutMs = deadlineGenerator.getOperationTimeoutMs();
 
-    Deadline deadline = getOperationCallOptions().getDeadline();
-    if (deadline != null) {
-      timeoutMs = deadline.timeRemaining(TimeUnit.MILLISECONDS);
-    }
+    // TODO(stepanian): Consider if the maxElapsedBackoffMillis logic could/should be included in
+    // DeadlineGenerator.
+    long timeoutMs = operationTimeoutMs.or((long) retryOptions.getMaxElapsedBackoffMillis());
 
     RetrySettings retrySettings =
         RetrySettings.newBuilder()
@@ -377,58 +377,18 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
               ImmutableMap.of("attempt", AttributeValue.longAttributeValue(failedCount))));
       Metadata metadata = new Metadata();
       metadata.merge(originalMetadata);
-      callWrapper.setCallAndStart(rpc, getRpcCallOptions(), getRetryRequest(), this, metadata);
+
+      Optional<Deadline> rpcAttemptDeadline = deadlineGenerator.getRpcAttemptDeadline();
+      CallOptions callOptions =
+          rpcAttemptDeadline.isPresent()
+              ? CallOptions.DEFAULT.withDeadline(rpcAttemptDeadline.get())
+              : CallOptions.DEFAULT;
+      callWrapper.setCallAndStart(rpc, callOptions, getRetryRequest(), this, metadata);
     } catch (Exception e) {
       setException(e);
     }
   }
 
-  /**
-   * Returns the {@link CallOptions} that a user set for the entire Operation, which can span
-   * multiple RPCs/retries.
-   *
-   * @return The {@link CallOptions}
-   */
-  protected CallOptions getOperationCallOptions() {
-    return callOptions;
-  }
-
-  /**
-   * Create an {@link CallOptions} that has a fail safe RPC deadline to make sure that unary
-   * operations don't hang. This will have to be overridden for streaming RPCs like read rows.
-   *
-   * <p>The logic is as follows:
-   *
-   * <ol>
-   *   <li>If the user provides a deadline, use the deadline
-   *   <li>Else If this is a streaming read, don't set an explicit deadline. The {@link
-   *       com.google.cloud.bigtable.grpc.io.Watchdog} will handle hanging
-   *   <li>Else Set a deadline of {@link #UNARY_DEADLINE_MINUTES} minutes deadline.
-   * </ol>
-   *
-   * @see com.google.cloud.bigtable.grpc.io.Watchdog Watchdog which handles hanging for streaming
-   *     reads.
-   * @return a {@link CallOptions}
-   */
-  protected CallOptions getRpcCallOptions() {
-    if (callOptions.getDeadline() != null || isStreamingRead()) {
-      // If the user set a deadline, honor it.
-      // If this is a streaming read, then the Watchdog will take affect and ensure that hanging
-      // does not occur.
-      return getOperationCallOptions();
-    } else {
-      // Unary calls should fail after 6 minutes, if there isn't any response from the server.
-      return callOptions.withDeadlineAfter(UNARY_DEADLINE_MINUTES, TimeUnit.MINUTES);
-    }
-  }
-
-  // TODO(sduskis): This is only required because BigtableDataGrpcClient doesn't always use
-  //      RetryingReadRowsOperation like it should.
-  protected boolean isStreamingRead() {
-    return request instanceof ReadRowsRequest
-        && !CallOptionsFactory.ConfiguredCallOptionsFactory.isGet((ReadRowsRequest) request);
-  }
-
   protected RequestT getRetryRequest() {
     return request;
   }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingMutateRowsOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingMutateRowsOperation.java
index 90975d7cd..51b21851d 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingMutateRowsOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingMutateRowsOperation.java
@@ -21,9 +21,9 @@ import com.google.bigtable.v2.MutateRowsRequest;
 import com.google.bigtable.v2.MutateRowsResponse;
 import com.google.cloud.bigtable.config.RetryOptions;
 import com.google.cloud.bigtable.grpc.BigtableDataClient;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
 import com.google.cloud.bigtable.grpc.async.MutateRowsRequestManager.ProcessingStatus;
 import com.google.common.collect.ImmutableMap;
-import io.grpc.CallOptions;
 import io.grpc.Metadata;
 import io.opencensus.trace.AttributeValue;
 import java.util.Arrays;
@@ -48,7 +48,7 @@ public class RetryingMutateRowsOperation
       RetryOptions retryOptions,
       MutateRowsRequest originalRequest,
       BigtableAsyncRpc<MutateRowsRequest, MutateRowsResponse> retryableRpc,
-      CallOptions callOptions,
+      DeadlineGenerator deadlineGenerator,
       ScheduledExecutorService retryExecutorService,
       Metadata originalMetadata,
       ApiClock clock) {
@@ -56,7 +56,7 @@ public class RetryingMutateRowsOperation
         retryOptions,
         originalRequest,
         retryableRpc,
-        callOptions,
+        deadlineGenerator,
         retryExecutorService,
         originalMetadata,
         clock);
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java
index f13c76def..e4e1b9922 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java
@@ -18,8 +18,8 @@ package com.google.cloud.bigtable.grpc.async;
 import com.google.api.core.ApiClock;
 import com.google.api.core.InternalApi;
 import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
 import com.google.common.collect.ImmutableList;
-import io.grpc.CallOptions;
 import io.grpc.Metadata;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
@@ -42,7 +42,7 @@ public class RetryingStreamOperation<RequestT, ResponseT>
    * @param retryOptions a {@link com.google.cloud.bigtable.config.RetryOptions} object.
    * @param request a RequestT object.
    * @param retryableRpc a {@link com.google.cloud.bigtable.grpc.async.BigtableAsyncRpc} object.
-   * @param callOptions a {@link io.grpc.CallOptions} object.
+   * @param deadlineGenerator a {@link DeadlineGenerator} object.
    * @param executorService a {@link java.util.concurrent.ScheduledExecutorService} object.
    * @param metadata a {@link io.grpc.Metadata} object.
    * @param clock a {@link ApiClock} object
@@ -51,11 +51,11 @@ public class RetryingStreamOperation<RequestT, ResponseT>
       RetryOptions retryOptions,
       RequestT request,
       BigtableAsyncRpc<RequestT, ResponseT> retryableRpc,
-      CallOptions callOptions,
+      DeadlineGenerator deadlineGenerator,
       ScheduledExecutorService executorService,
       Metadata metadata,
       ApiClock clock) {
-    super(retryOptions, request, retryableRpc, callOptions, executorService, metadata, clock);
+    super(retryOptions, request, retryableRpc, deadlineGenerator, executorService, metadata, clock);
   }
 
   /** {@inheritDoc} */
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingUnaryOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingUnaryOperation.java
index 2136cd435..d16187e66 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingUnaryOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingUnaryOperation.java
@@ -18,7 +18,7 @@ package com.google.cloud.bigtable.grpc.async;
 import com.google.api.core.ApiClock;
 import com.google.api.core.InternalApi;
 import com.google.cloud.bigtable.config.RetryOptions;
-import io.grpc.CallOptions;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
@@ -43,7 +43,7 @@ public class RetryingUnaryOperation<RequestT, ResponseT>
    * @param retryOptions a {@link com.google.cloud.bigtable.config.RetryOptions} object.
    * @param request a RequestT object.
    * @param retryableRpc a {@link com.google.cloud.bigtable.grpc.async.BigtableAsyncRpc} object.
-   * @param callOptions a {@link io.grpc.CallOptions} object.
+   * @param deadlineGenerator a {@link DeadlineGenerator} object.
    * @param executorService a {@link java.util.concurrent.ScheduledExecutorService} object.
    * @param metadata a {@link io.grpc.Metadata} object.
    * @param clock a {@link ApiClock} object
@@ -52,11 +52,11 @@ public class RetryingUnaryOperation<RequestT, ResponseT>
       RetryOptions retryOptions,
       RequestT request,
       BigtableAsyncRpc<RequestT, ResponseT> retryableRpc,
-      CallOptions callOptions,
+      DeadlineGenerator deadlineGenerator,
       ScheduledExecutorService executorService,
       Metadata metadata,
       ApiClock clock) {
-    super(retryOptions, request, retryableRpc, callOptions, executorService, metadata, clock);
+    super(retryOptions, request, retryableRpc, deadlineGenerator, executorService, metadata, clock);
   }
 
   /** {@inheritDoc} */
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
index fcb6abb61..cb779c751 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
@@ -21,6 +21,7 @@ import com.google.api.core.InternalApi;
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.ReadRowsResponse;
 import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
 import com.google.cloud.bigtable.grpc.async.AbstractRetryingOperation;
 import com.google.cloud.bigtable.grpc.async.BigtableAsyncRpc;
 import com.google.cloud.bigtable.grpc.async.CallController;
@@ -28,7 +29,6 @@ import com.google.cloud.bigtable.grpc.io.Watchdog.StreamWaitTimeoutException;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.ByteString;
-import io.grpc.CallOptions;
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.stub.ClientResponseObserver;
@@ -84,7 +84,7 @@ public class RetryingReadRowsOperation
       RetryOptions retryOptions,
       ReadRowsRequest request,
       BigtableAsyncRpc<ReadRowsRequest, ReadRowsResponse> retryableRpc,
-      CallOptions callOptions,
+      DeadlineGenerator deadlineGenerator,
       ScheduledExecutorService retryExecutorService,
       Metadata originalMetadata,
       ApiClock clock) {
@@ -92,7 +92,7 @@ public class RetryingReadRowsOperation
         retryOptions,
         request,
         retryableRpc,
-        callOptions,
+        deadlineGenerator,
         retryExecutorService,
         originalMetadata,
         clock);
diff --git a/bigtable-client-core-parent/bigtable-hbase/pom.xml b/bigtable-client-core-parent/bigtable-hbase/pom.xml
index d680ee965..9665074aa 100644
--- a/bigtable-client-core-parent/bigtable-hbase/pom.xml
+++ b/bigtable-client-core-parent/bigtable-hbase/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase</artifactId>
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
index 31ad40481..e78071a58 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
@@ -25,6 +25,7 @@ import static com.google.cloud.bigtable.config.CallOptionsConfig.SHORT_TIMEOUT_M
 import static com.google.cloud.bigtable.config.CallOptionsConfig.USE_TIMEOUT_DEFAULT;
 import static com.google.common.base.Strings.isNullOrEmpty;
 
+import com.google.api.core.BetaApi;
 import com.google.api.core.InternalExtensionOnly;
 import com.google.auth.Credentials;
 import com.google.cloud.bigtable.config.BigtableOptions;
@@ -33,6 +34,7 @@ import com.google.cloud.bigtable.config.CallOptionsConfig;
 import com.google.cloud.bigtable.config.CredentialOptions;
 import com.google.cloud.bigtable.config.Logger;
 import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import io.grpc.Status;
 import java.io.FileInputStream;
@@ -264,6 +266,16 @@ public class BigtableOptionsFactory {
    */
   public static final String BIGTABLE_RPC_TIMEOUT_MS_KEY = "google.bigtable.rpc.timeout.ms";
 
+  /**
+   * If timeouts are set, how many milliseconds should pass before a DEADLINE_EXCEEDED for an
+   * individual RPC attempt? Note that multiple attempts may happen within an overall operation,
+   * whose timeout is governed by {@link #BIGTABLE_RPC_TIMEOUT_MS_KEY}. Currently, this feature is
+   * experimental.
+   */
+  @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+  public static final String BIGTABLE_RPC_ATTEMPT_TIMEOUT_MS_KEY =
+      "google.bigtable.rpc.attempt.timeout.ms";
+
   /**
    * If timeouts are set, how many milliseconds should pass before a DEADLINE_EXCEEDED for a long
    * read? Currently, this feature is experimental.
@@ -282,6 +294,16 @@ public class BigtableOptionsFactory {
   public static final String BIGTABLE_MUTATE_RPC_TIMEOUT_MS_KEY =
       "google.bigtable.mutate.rpc.timeout.ms";
 
+  /**
+   * If timeouts are set, how many milliseconds should pass before a DEADLINE_EXCEEDED for an RPC
+   * attempt within a long mutation. Note that multiple attempts may happen within an overall
+   * operation, whose timeout is governed by {@link #BIGTABLE_MUTATE_RPC_TIMEOUT_MS_KEY}. Currently,
+   * this feature is experimental.
+   */
+  @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+  public static final String BIGTABLE_MUTATE_RPC_ATTEMPT_TIMEOUT_MS_KEY =
+      "google.bigtable.mutate.rpc.attempt.timeout.ms";
+
   /**
    * If timeouts are set, how many milliseconds should pass before a DEADLINE_EXCEEDED for a long
    * read. Currently, this feature is experimental.
@@ -289,6 +311,16 @@ public class BigtableOptionsFactory {
   public static final String BIGTABLE_READ_RPC_TIMEOUT_MS_KEY =
       "google.bigtable.read.rpc.timeout.ms";
 
+  /**
+   * If timeouts are set, how many milliseconds should pass before a DEADLINE_EXCEEDED for an RPC
+   * attempt within a long mutation. Note that multiple attempts may happen within an overall
+   * operation, whose timeout is governed by {@link #BIGTABLE_READ_RPC_TIMEOUT_MS_KEY}. Currently,
+   * this feature is experimental.
+   */
+  @BetaApi("The API for setting attempt timeouts is not yet stable and may change in the future")
+  public static final String BIGTABLE_READ_RPC_ATTEMPT_TIMEOUT_MS_KEY =
+      "google.bigtable.read.rpc.attempt.timeout.ms";
+
   /** Allow namespace methods to be no-ops */
   public static final String BIGTABLE_NAMESPACE_WARNING_KEY = "google.bigtable.namespace.warnings";
 
@@ -493,15 +525,45 @@ public class BigtableOptionsFactory {
         configuration.getBoolean(BIGTABLE_USE_TIMEOUTS_KEY, USE_TIMEOUT_DEFAULT));
     clientCallOptionsBuilder.setShortRpcTimeoutMs(
         configuration.getInt(BIGTABLE_RPC_TIMEOUT_MS_KEY, SHORT_TIMEOUT_MS_DEFAULT));
+
+    Optional<Integer> rpcAttemptTimeoutMs =
+        getOptionalIntConfig(configuration, BIGTABLE_RPC_ATTEMPT_TIMEOUT_MS_KEY);
+    if (rpcAttemptTimeoutMs.isPresent()) {
+      clientCallOptionsBuilder.setShortRpcAttemptTimeoutMs(rpcAttemptTimeoutMs.get());
+    }
+
     int longTimeoutMs =
         configuration.getInt(BIGTABLE_LONG_RPC_TIMEOUT_MS_KEY, LONG_TIMEOUT_MS_DEFAULT);
     clientCallOptionsBuilder.setMutateRpcTimeoutMs(
         configuration.getInt(BIGTABLE_MUTATE_RPC_TIMEOUT_MS_KEY, longTimeoutMs));
+
+    Optional<Integer> mutateRpcAttemptTimeoutMs =
+        getOptionalIntConfig(configuration, BIGTABLE_MUTATE_RPC_ATTEMPT_TIMEOUT_MS_KEY);
+    if (mutateRpcAttemptTimeoutMs.isPresent()) {
+      clientCallOptionsBuilder.setMutateRpcAttemptTimeoutMs(mutateRpcAttemptTimeoutMs.get());
+    }
+
     clientCallOptionsBuilder.setReadRowsRpcTimeoutMs(
         configuration.getInt(BIGTABLE_READ_RPC_TIMEOUT_MS_KEY, longTimeoutMs));
+
+    Optional<Integer> readRpcAttemptTimeoutMs =
+        getOptionalIntConfig(configuration, BIGTABLE_READ_RPC_ATTEMPT_TIMEOUT_MS_KEY);
+    if (readRpcAttemptTimeoutMs.isPresent()) {
+      clientCallOptionsBuilder.setReadRowsRpcAttemptTimeoutMs(readRpcAttemptTimeoutMs.get());
+    }
     bigtableOptionsBuilder.setCallOptionsConfig(clientCallOptionsBuilder.build());
   }
 
+  private static Optional<Integer> getOptionalIntConfig(
+      Configuration configuration, String property) {
+    String value = configuration.getTrimmed(property);
+    if (value != null) {
+      return Optional.of(Integer.parseInt(value));
+    } else {
+      return Optional.absent();
+    }
+  }
+
   private static RetryOptions createRetryOptions(Configuration configuration) {
     RetryOptions.Builder retryOptionsBuilder = RetryOptions.builder();
     boolean enableRetries =
diff --git a/bigtable-client-core-parent/pom.xml b/bigtable-client-core-parent/pom.xml
index b7a8fa041..cf46758c4 100644
--- a/bigtable-client-core-parent/pom.xml
+++ b/bigtable-client-core-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   </parent>
 
   <artifactId>bigtable-client-core-parent</artifactId>
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
index e017befcb..1a13f872f 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
+++ b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-dataflow-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml b/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
index df4bc06e0..038bc6ab4 100755
--- a/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
+++ b/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
@@ -16,7 +16,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-dataflow-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-beam</artifactId>
diff --git a/bigtable-dataflow-parent/pom.xml b/bigtable-dataflow-parent/pom.xml
index 704a77f5d..6c9f17c49 100644
--- a/bigtable-dataflow-parent/pom.xml
+++ b/bigtable-dataflow-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
   </parent>
 
   <artifactId>bigtable-dataflow-parent</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
index d6aada089..3d4e83367 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
@@ -3,7 +3,7 @@
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
index 12c0d4066..b7bf5db7f 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
index 7015e09bc..ca35df048 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x-mapreduce</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
index aca13bf2c..1b1a936d2 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
index e4bfa2f4f..5dc3aa6dc 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x</artifactId>
diff --git a/bigtable-hbase-1.x-parent/pom.xml b/bigtable-hbase-1.x-parent/pom.xml
index d176a5ef7..3a0ca1269 100644
--- a/bigtable-hbase-1.x-parent/pom.xml
+++ b/bigtable-hbase-1.x-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x-parent</artifactId>
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
index 6bed8bba4..1621024aa 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
index 636437cf5..8e8dc5e07 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
index c997dd354..66f9b169e 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
 
   <properties>
diff --git a/bigtable-hbase-2.x-parent/pom.xml b/bigtable-hbase-2.x-parent/pom.xml
index 2689f2720..0d8a968c8 100644
--- a/bigtable-hbase-2.x-parent/pom.xml
+++ b/bigtable-hbase-2.x-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-2.x-parent</artifactId>
diff --git a/pom.xml b/pom.xml
index 31c67d031..63491a852 100644
--- a/pom.xml
+++ b/pom.xml
@@ -19,7 +19,7 @@ limitations under the License.
 
   <groupId>com.google.cloud.bigtable</groupId>
   <artifactId>bigtable-client-parent</artifactId>
-  <version>1.17.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+  <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   <packaging>pom</packaging>
   <name>${project.groupId}:${project.artifactId}</name>
   <url>https://cloud.google.com/bigtable/</url>
@@ -146,6 +146,11 @@ limitations under the License.
             <compilerArgument>-Xlint:-options</compilerArgument>
           </configuration>
         </plugin>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-dependency-plugin</artifactId>
+          <version>3.1.2</version>
+        </plugin>
         <plugin>
           <groupId>org.apache.maven.plugins</groupId>
           <artifactId>maven-surefire-plugin</artifactId>
@@ -250,6 +255,11 @@ limitations under the License.
           <style>google</style>
         </configuration>
       </plugin>
+      <!-- Define dependency-plugin for convenience, notably to easily check dependency tree from IntelliJ. -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+      </plugin>
     </plugins>
   </build>
 </project>
diff --git a/versions.txt b/versions.txt
index 36310bba3..9d49fe128 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,9 +1,9 @@
 # Format:
 # module:released-version:current-version
 
-bigtable-client-parent:1.17.1:1.17.1
-bigtable-client-core-parent:1.17.1:1.17.1
-bigtable-hbase-1x-parent:1.17.1:1.17.1
-bigtable-hbase-2x-parent:1.17.1:1.17.1
-bigtable-dataflow-parent:1.17.1:1.17.1
-bigtable-test:1.17.1:1.17.1
+bigtable-client-parent:1.17.1:1.17.2-SNAPSHOT
+bigtable-client-core-parent:1.17.1:1.17.2-SNAPSHOT
+bigtable-hbase-1x-parent:1.17.1:1.17.2-SNAPSHOT
+bigtable-hbase-2x-parent:1.17.1:1.17.2-SNAPSHOT
+bigtable-dataflow-parent:1.17.1:1.17.2-SNAPSHOT
+bigtable-test:1.17.1:1.17.2-SNAPSHOT


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestCallOptionsFactory.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestDeadlineGeneratorFactory.java
similarity index 68%
rename from bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestCallOptionsFactory.java
rename to bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestDeadlineGeneratorFactory.java
index e0ad8c2dc..b17fbc637 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestCallOptionsFactory.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestDeadlineGeneratorFactory.java
@@ -15,12 +15,18 @@
  */
 package com.google.cloud.bigtable.grpc;
 
+import static org.hamcrest.CoreMatchers.both;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
 import com.google.bigtable.v2.MutateRowRequest;
 import com.google.bigtable.v2.MutateRowsRequest;
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.SampleRowKeysRequest;
 import com.google.cloud.bigtable.config.CallOptionsConfig;
 import com.google.cloud.bigtable.grpc.async.OperationClock;
+import com.google.common.base.Optional;
 import io.grpc.CallOptions;
 import io.grpc.Context;
 import io.grpc.Deadline;
@@ -36,19 +42,41 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
 
-/** Tests for {@link CallOptionsFactory}. */
+/** Tests for {@link DeadlineGeneratorFactory}. */
 @RunWith(JUnit4.class)
-public class TestCallOptionsFactory {
+public class TestDeadlineGeneratorFactory {
   @Rule public final MockitoRule mockitoRule = MockitoJUnit.rule();
 
   private static final int LARGE_TIMEOUT = 500_000;
 
+  /**
+   * Utility method to return a mock RequestCallOptions that answers all calls to {@link
+   * DeadlineGenerator#getRpcAttemptDeadline()} with the given callOptions object.
+   */
+  public static DeadlineGenerator mockCallOptionsFactory(final CallOptions callOptions) {
+    return new DeadlineGenerator() {
+      @Override
+      public Optional<Long> getOperationTimeoutMs() {
+        Deadline deadline = callOptions.getDeadline();
+        return deadline != null
+            ? Optional.of(deadline.timeRemaining(TimeUnit.MILLISECONDS))
+            : Optional.<Long>absent();
+      }
+
+      @Override
+      public Optional<Deadline> getRpcAttemptDeadline() {
+        return Optional.fromNullable(callOptions.getDeadline());
+      }
+    };
+  }
+
   @Mock ScheduledExecutorService mockExecutor;
 
   @Test
   public void testDefault() {
-    CallOptionsFactory factory = new CallOptionsFactory.Default();
-    Assert.assertSame(CallOptions.DEFAULT, factory.create(null, null));
+    DeadlineGeneratorFactory factory = DeadlineGeneratorFactory.DEFAULT;
+    Assert.assertFalse(
+        factory.getRequestDeadlineGenerator(null).getRpcAttemptDeadline().isPresent());
   }
 
   @Test
@@ -60,7 +88,7 @@ public class TestCallOptionsFactory {
         new Runnable() {
           @Override
           public void run() {
-            CallOptionsFactory factory = new CallOptionsFactory.Default();
+            DeadlineGeneratorFactory factory = DeadlineGeneratorFactory.DEFAULT;
             assertEqualsDeadlines(
                 deadline.timeRemaining(TimeUnit.MILLISECONDS), getDeadlineMs(factory, null));
           }
@@ -70,8 +98,15 @@ public class TestCallOptionsFactory {
   @Test
   public void testConfiguredDefaultConfig() {
     CallOptionsConfig config = CallOptionsConfig.builder().build();
-    CallOptionsFactory factory = new CallOptionsFactory.ConfiguredCallOptionsFactory(config);
-    Assert.assertSame(CallOptions.DEFAULT, factory.create(null, null));
+    DeadlineGeneratorFactory factory = new ConfiguredDeadlineGeneratorFactory(config);
+
+    assertThat(
+        factory
+            .getRequestDeadlineGenerator(null)
+            .getRpcAttemptDeadline()
+            .get()
+            .timeRemaining(TimeUnit.MINUTES),
+        both(greaterThanOrEqualTo(5L)).and(lessThanOrEqualTo(7L)));
   }
 
   @Test
@@ -81,7 +116,7 @@ public class TestCallOptionsFactory {
             .setUseTimeout(true)
             .setReadRowsRpcTimeoutMs(LARGE_TIMEOUT)
             .build();
-    CallOptionsFactory factory = new CallOptionsFactory.ConfiguredCallOptionsFactory(config);
+    DeadlineGeneratorFactory factory = new ConfiguredDeadlineGeneratorFactory(config);
     assertEqualsDeadlines(
         config.getShortRpcTimeoutMs(),
         getDeadlineMs(factory, SampleRowKeysRequest.getDefaultInstance()));
@@ -108,8 +143,7 @@ public class TestCallOptionsFactory {
                     .setMutateRpcTimeoutMs((int) TimeUnit.SECONDS.toMillis(1000))
                     .setReadRowsRpcTimeoutMs((int) TimeUnit.SECONDS.toMillis(1000))
                     .build();
-            CallOptionsFactory factory =
-                new CallOptionsFactory.ConfiguredCallOptionsFactory(config);
+            DeadlineGeneratorFactory factory = new ConfiguredDeadlineGeneratorFactory(config);
             // The deadline in the context in 1 second, and the deadline in the config is 100+
             // seconds
             assertEqualsDeadlines(
@@ -131,7 +165,7 @@ public class TestCallOptionsFactory {
             .setMutateRpcTimeoutMs(mutate_time)
             .setReadRowsRpcTimeoutMs(read_time)
             .build();
-    CallOptionsFactory factory = new CallOptionsFactory.ConfiguredCallOptionsFactory(config);
+    DeadlineGeneratorFactory factory = new ConfiguredDeadlineGeneratorFactory(config);
     // The deadline in the context in 1 second, and the deadline in the config is 100+
     // seconds
     assertEqualsDeadlines(
@@ -148,7 +182,7 @@ public class TestCallOptionsFactory {
             .setUseTimeout(true)
             .setMutateRpcTimeoutMs(LARGE_TIMEOUT)
             .build();
-    CallOptionsFactory factory = new CallOptionsFactory.ConfiguredCallOptionsFactory(config);
+    DeadlineGeneratorFactory factory = new ConfiguredDeadlineGeneratorFactory(config);
     assertEqualsDeadlines(
         LARGE_TIMEOUT, getDeadlineMs(factory, MutateRowsRequest.getDefaultInstance()));
   }
@@ -156,9 +190,12 @@ public class TestCallOptionsFactory {
   @Test
   public void testWhenNoTimeout() {
     CallOptionsConfig config = CallOptionsConfig.builder().setUseTimeout(false).build();
-    CallOptionsFactory factory = new CallOptionsFactory.ConfiguredCallOptionsFactory(config);
-    Assert.assertEquals(
-        CallOptions.DEFAULT, factory.create(null, ReadRowsRequest.getDefaultInstance()));
+    DeadlineGeneratorFactory factory = new ConfiguredDeadlineGeneratorFactory(config);
+    Assert.assertFalse(
+        factory
+            .getRequestDeadlineGenerator(ReadRowsRequest.getDefaultInstance())
+            .getRpcAttemptDeadline()
+            .isPresent());
   }
 
   /**
@@ -172,7 +209,12 @@ public class TestCallOptionsFactory {
     Assert.assertEquals((double) expected, (double) actual, 10);
   }
 
-  private int getDeadlineMs(CallOptionsFactory factory, Object request) {
-    return (int) factory.create(null, request).getDeadline().timeRemaining(TimeUnit.MILLISECONDS);
+  private int getDeadlineMs(DeadlineGeneratorFactory factory, Object request) {
+    return (int)
+        factory
+            .getRequestDeadlineGenerator(request)
+            .getRpcAttemptDeadline()
+            .get()
+            .timeRemaining(TimeUnit.MILLISECONDS);
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingMutateRowsOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingMutateRowsOperation.java
index 419c1dc98..edb8d9087 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingMutateRowsOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingMutateRowsOperation.java
@@ -29,9 +29,9 @@ import com.google.bigtable.v2.MutateRowsResponse;
 import com.google.bigtable.v2.Mutation;
 import com.google.bigtable.v2.Mutation.SetCell;
 import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.rpc.Status;
-import io.grpc.CallOptions;
 import io.grpc.ClientCall;
 import io.grpc.Metadata;
 import io.grpc.Status.Code;
@@ -247,7 +247,7 @@ public class TestRetryingMutateRowsOperation {
         RETRY_OPTIONS,
         request,
         mutateRows,
-        CallOptions.DEFAULT,
+        DeadlineGenerator.DEFAULT,
         executorService,
         new Metadata(),
         clock);
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingUnaryOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingUnaryOperation.java
index 26c0c80dc..93def971e 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingUnaryOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestRetryingUnaryOperation.java
@@ -25,6 +25,8 @@ import com.google.bigtable.v2.BigtableGrpc;
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.ReadRowsResponse;
 import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
+import com.google.cloud.bigtable.grpc.TestDeadlineGeneratorFactory;
 import com.google.cloud.bigtable.grpc.scanner.BigtableRetriesExhaustedException;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.grpc.CallOptions;
@@ -97,7 +99,7 @@ public class TestRetryingUnaryOperation {
             (ClientCall.Listener) any(),
             (Metadata) any(),
             (ClientCall) any());
-    ListenableFuture future = createOperation(CallOptions.DEFAULT).getAsyncResult();
+    ListenableFuture future = createOperation(DeadlineGenerator.DEFAULT).getAsyncResult();
     Assert.assertEquals(result, future.get(1, TimeUnit.SECONDS));
     verify(readAsync, times(1))
         .start(
@@ -133,7 +135,7 @@ public class TestRetryingUnaryOperation {
             Mockito.<ClientCall.Listener>any(),
             Mockito.<Metadata>any(),
             Mockito.<ClientCall>any());
-    ListenableFuture future = createOperation(CallOptions.DEFAULT).getAsyncResult();
+    ListenableFuture future = createOperation(DeadlineGenerator.DEFAULT).getAsyncResult();
 
     Assert.assertEquals(result, future.get(1, TimeUnit.SECONDS));
     Assert.assertEquals(5, counter.get());
@@ -201,7 +203,9 @@ public class TestRetryingUnaryOperation {
         .when(readAsync)
         .start((ReadRowsRequest) any(), (Listener) any(), (Metadata) any(), (ClientCall) any());
     try {
-      createOperation(options).getAsyncResult().get(1, TimeUnit.SECONDS);
+      createOperation(TestDeadlineGeneratorFactory.mockCallOptionsFactory(options))
+          .getAsyncResult()
+          .get(1, TimeUnit.SECONDS);
       Assert.fail();
     } catch (ExecutionException e) {
       Assert.assertEquals(BigtableRetriesExhaustedException.class, e.getCause().getClass());
@@ -214,12 +218,12 @@ public class TestRetryingUnaryOperation {
     clock.assertTimeWithinExpectations(TimeUnit.MILLISECONDS.toNanos(expectedTimeoutMs));
   }
 
-  private RetryingUnaryOperation createOperation(CallOptions options) {
+  private RetryingUnaryOperation createOperation(DeadlineGenerator deadlineGenerator) {
     return new RetryingUnaryOperation<>(
         RETRY_OPTIONS,
         ReadRowsRequest.getDefaultInstance(),
         readAsync,
-        options,
+        deadlineGenerator,
         executorService,
         new Metadata(),
         clock);
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
index f1594affd..2cd752a6d 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperationTest.java
@@ -32,6 +32,8 @@ import com.google.bigtable.v2.ReadRowsResponse.CellChunk;
 import com.google.bigtable.v2.RowRange;
 import com.google.bigtable.v2.RowSet;
 import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.cloud.bigtable.grpc.DeadlineGenerator;
+import com.google.cloud.bigtable.grpc.TestDeadlineGeneratorFactory;
 import com.google.cloud.bigtable.grpc.async.BigtableAsyncRpc;
 import com.google.cloud.bigtable.grpc.async.BigtableAsyncRpc.RpcMetrics;
 import com.google.cloud.bigtable.grpc.async.OperationClock;
@@ -140,18 +142,18 @@ public class RetryingReadRowsOperationTest {
   }
 
   protected RetryingReadRowsOperation createOperation() {
-    return createOperation(CallOptions.DEFAULT, mockFlatRowObserver);
+    return createOperation(DeadlineGenerator.DEFAULT, mockFlatRowObserver);
   }
 
   protected RetryingReadRowsOperation createOperation(
-      CallOptions options, StreamObserver<FlatRow> observer) {
+      DeadlineGenerator deadlineGenerator, StreamObserver<FlatRow> observer) {
     RetryingReadRowsOperation operation =
         new RetryingReadRowsOperation(
             observer,
             RETRY_OPTIONS,
             READ_ENTIRE_TABLE_REQUEST,
             mockRetryableRpc,
-            options,
+            deadlineGenerator,
             mockRetryExecutorService,
             metaData,
             clock);
@@ -221,16 +223,18 @@ public class RetryingReadRowsOperationTest {
 
   @Test
   public void testFailure_default() throws Exception {
-    testFailure(RETRY_OPTIONS.getMaxElapsedBackoffMillis(), CallOptions.DEFAULT);
+    testFailure(RETRY_OPTIONS.getMaxElapsedBackoffMillis(), DeadlineGenerator.DEFAULT);
   }
 
   @Test
   public void testFailure_deadline() throws Exception {
-    CallOptions options = DeadlineUtil.optionsWithDeadline(1, TimeUnit.SECONDS, clock);
-    testFailure(TimeUnit.SECONDS.toMillis(1), options);
+    DeadlineGenerator deadlineGenerator =
+        TestDeadlineGeneratorFactory.mockCallOptionsFactory(
+            DeadlineUtil.optionsWithDeadline(1, TimeUnit.SECONDS, clock));
+    testFailure(TimeUnit.SECONDS.toMillis(1), deadlineGenerator);
   }
 
-  private void testFailure(long expectedTimeMs, CallOptions options)
+  private void testFailure(long expectedTimeMs, DeadlineGenerator deadlineGenerator)
       throws InterruptedException, java.util.concurrent.TimeoutException {
     doAnswer(
             new Answer<Void>() {
@@ -249,7 +253,7 @@ public class RetryingReadRowsOperationTest {
             any(Metadata.class),
             any(ClientCall.class));
 
-    RetryingReadRowsOperation underTest = createOperation(options, mockFlatRowObserver);
+    RetryingReadRowsOperation underTest = createOperation(deadlineGenerator, mockFlatRowObserver);
     try {
       underTest.getAsyncResult().get(100, TimeUnit.MILLISECONDS);
     } catch (ExecutionException e) {
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
index c1343a28e..df7308c28 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
@@ -19,7 +19,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-integration-tests-common</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
index 91089d8d7..560883749 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x-integration-tests</artifactId>
@@ -372,6 +372,12 @@ limitations under the License.
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>com.google.api.grpc</groupId>
+      <artifactId>grpc-google-cloud-bigtable-v2</artifactId>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-bigtable-v2</artifactId>
@@ -452,6 +458,12 @@ limitations under the License.
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-stub</artifactId>
+      <scope>test</scope>
+    </dependency>
+
 
     <!-- Testing deps -->
     <dependency>
@@ -460,6 +472,12 @@ limitations under the License.
       <version>${hamcrest.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-library</artifactId>
+      <version>${hamcrest.version}</version>
+      <scope>test</scope>
+    </dependency>
 
     <dependency>
       <groupId>junit</groupId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java
index eac6045e0..8539b1474 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java
@@ -43,6 +43,11 @@ import org.junit.runners.Suite;
   TestIncrement.class,
   TestListTables.class,
   TestPut.class,
+  TestRpcRetryBehaviorCheckAndMutateRow.class,
+  TestRpcRetryBehaviorGetMultiple.class,
+  TestRpcRetryBehaviorGetSingle.class,
+  TestRpcRetryBehaviorPutMultiple.class,
+  TestRpcRetryBehaviorPutSingle.class,
   TestTimestamp.class,
   TestTruncateTable.class,
   TestModifyTable.class,
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehavior.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehavior.java
new file mode 100644
index 000000000..556090db5
--- /dev/null
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehavior.java
@@ -0,0 +1,279 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.hbase;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThan;
+
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.CheckAndMutateRowRequest;
+import com.google.bigtable.v2.CheckAndMutateRowResponse;
+import com.google.bigtable.v2.MutateRowRequest;
+import com.google.bigtable.v2.MutateRowResponse;
+import com.google.bigtable.v2.MutateRowsRequest;
+import com.google.bigtable.v2.MutateRowsResponse;
+import com.google.bigtable.v2.ReadModifyWriteRowRequest;
+import com.google.bigtable.v2.ReadModifyWriteRowResponse;
+import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.bigtable.v2.ReadRowsResponse;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang.time.StopWatch;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Table;
+import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(KnownHBaseGap.class)
+public abstract class TestRpcRetryBehavior {
+  /**
+   * Indicates whether the test will involve timeouts from the user side. In case we don't, the
+   * client would only retry the logic on a server response.
+   */
+  protected boolean timeoutEnabled;
+  /**
+   * Indicates whether the test is for attempt timeouts enabled (the new opt-in feature as of
+   * 1.18.x) or disabled (the default logic for the versions prior to that).
+   */
+  protected boolean attemptTimeoutEnabled;
+  /**
+   * Indicates whether our test server RPC will explicitly abort (true) or hang (false). This allows
+   * us to test the timeout capability on the client side.
+   */
+  protected boolean serverRpcAbortsForTest;
+
+  // Choose timeouts such that we can have multiple attempts while having some overlap in the last
+  // attempt over the timeout. Here, we have attempts at the approximate times 0, 450, 900, 1350,
+  // 1800 ms before reaching timeout at 2000 ms. Hence, we expect 5 attempts. The last attempt at
+  // 1800 would end at 2250 if it weren't for a timeout, so our tests need to verify that our
+  // operations don't reach to that point and instead end at around 2000 ms.
+  protected final long attemptTimeoutMs = 450;
+  protected final long operationTimeoutMs = 2000;
+  protected final int expectedAttemptsWithRetryLogic = 5;
+  protected final long maxElapsedBackoffMs = 1900;
+
+  private Server server;
+
+  @Before
+  public void setup() throws Exception {
+    this.server = startFake();
+  }
+
+  @After
+  public void teardown() {
+    this.server.shutdownNow();
+  }
+
+  // TODO(stepanian): move away from these parameterized-test flags and move the logic directly into
+  // these test methods and out of testMain.
+  @Test
+  public void testOperationAndAttemptTimeoutsWillRetryOnServerAbort() throws Exception {
+    timeoutEnabled = true;
+    attemptTimeoutEnabled = true;
+    serverRpcAbortsForTest = true;
+    testMain();
+  }
+
+  @Test
+  public void testOperationAndAttemptTimeoutsWillRetryOnServerHang() throws Exception {
+    timeoutEnabled = true;
+    attemptTimeoutEnabled = true;
+    serverRpcAbortsForTest = false;
+    testMain();
+  }
+
+  @Test
+  public void testOperationTimeoutOnlyWillNotRetryAndEndEarlyOnServerAbort() throws Exception {
+    timeoutEnabled = true;
+    attemptTimeoutEnabled = false;
+    serverRpcAbortsForTest = true;
+    testMain();
+  }
+
+  @Test
+  public void testOperationTimeoutOnlyWillNotRetryAndEndEarlyOnServerHang() throws Exception {
+    timeoutEnabled = true;
+    attemptTimeoutEnabled = false;
+    serverRpcAbortsForTest = false;
+    testMain();
+  }
+
+  @Test
+  public void testNoTimeoutsWillStillRetryOnServerAbort() throws Exception {
+    timeoutEnabled = false;
+    attemptTimeoutEnabled = false;
+    serverRpcAbortsForTest = true;
+    testMain();
+  }
+
+  private void testMain() throws Exception {
+    ImmutableMap.Builder<String, String> connPropsBuilder = defineProperties();
+
+    try (Connection conn = makeConnection(connPropsBuilder.build())) {
+      Table table = conn.getTable(TableName.valueOf("table"));
+
+      StopWatch sw = new StopWatch();
+      executeLogic(table, sw);
+
+      validateOperationRuntime(sw);
+    }
+
+    validateInvocations(getInvocations());
+  }
+
+  protected abstract AtomicInteger getInvocations();
+
+  protected abstract void executeLogic(Table table, StopWatch sw) throws Exception;
+
+  protected abstract ImmutableMap.Builder<String, String> defineProperties();
+
+  protected void validateOperationRuntime(StopWatch sw) {
+    if (timeoutEnabled) {
+      assertThat(
+          sw.getTime(),
+          Matchers.both(greaterThanOrEqualTo(operationTimeoutMs))
+              .and(lessThan(operationTimeoutMs + 200))); // plus some buffer
+    } else {
+      // Without timeouts and just based on the backoff millis, the upper bound isn't as exact since
+      // we may or may not
+      // do an additional attempt. Update the upper bound accordingly.
+      assertThat(
+          sw.getTime(),
+          Matchers.both(greaterThanOrEqualTo(operationTimeoutMs))
+              .and(lessThan(operationTimeoutMs + attemptTimeoutMs + 200))); // plus some buffer
+    }
+  }
+
+  protected void validateInvocations(AtomicInteger counter) {
+    if (!timeoutEnabled && serverRpcAbortsForTest) {
+      // Backoff and jitter logic is not as deterministic. Best we can do is verify a minimum number
+      // of retries.
+      assertThat(counter.get(), greaterThanOrEqualTo(expectedAttemptsWithRetryLogic - 1));
+      return;
+    }
+
+    // Expect multiple retries (given our timeout parameters) if attempt timeouts are enabled;
+    // otherwise,
+    // only 1 retry happens.
+    int expectedInvocations;
+    if (attemptTimeoutEnabled || serverRpcAbortsForTest) {
+      expectedInvocations = expectedAttemptsWithRetryLogic;
+    } else {
+      expectedInvocations = 1;
+    }
+
+    assertThat(counter.get(), equalTo(expectedInvocations));
+  }
+
+  protected Connection makeConnection(Map<String, String> customConnProps) throws IOException {
+    Configuration config = BigtableConfiguration.configure("project", "instance");
+
+    config.set(BigtableOptionsFactory.BIGTABLE_DATA_CHANNEL_COUNT_KEY, "1");
+
+    // only for emulator
+    config.set(BigtableOptionsFactory.BIGTABLE_EMULATOR_HOST_KEY, "localhost:" + server.getPort());
+
+    for (Map.Entry<String, String> connProp : customConnProps.entrySet()) {
+      config.set(connProp.getKey(), connProp.getValue());
+    }
+
+    return ConnectionFactory.createConnection(config);
+  }
+
+  protected abstract BigtableGrpc.BigtableImplBase setupRpcCall();
+
+  private Server startFake() throws Exception {
+    final BigtableGrpc.BigtableImplBase rpcBase = setupRpcCall();
+    BigtableGrpc.BigtableImplBase rpcSleepWrapper = setupRpcServerWithSleepHandler(rpcBase);
+
+    int portNum = 54128; // Pick a port arbitrarily.
+    Server fakeBigtableServer = ServerBuilder.forPort(portNum).addService(rpcSleepWrapper).build();
+    fakeBigtableServer.start();
+    return fakeBigtableServer;
+  }
+
+  /**
+   * Wraps the given base RPC handler defined by the implementation tests with our handler for the
+   * sleep and abort logic.
+   */
+  private BigtableGrpc.BigtableImplBase setupRpcServerWithSleepHandler(
+      final BigtableGrpc.BigtableImplBase rpcBase) {
+    return new BigtableGrpc.BigtableImplBase() {
+      @Override
+      public void mutateRow(
+          MutateRowRequest request, StreamObserver<MutateRowResponse> responseObserver) {
+        rpcBase.mutateRow(request, responseObserver);
+        sleepAndAbortIfApplicable(responseObserver);
+      }
+
+      @Override
+      public void checkAndMutateRow(
+          CheckAndMutateRowRequest request,
+          StreamObserver<CheckAndMutateRowResponse> responseObserver) {
+        rpcBase.checkAndMutateRow(request, responseObserver);
+        sleepAndAbortIfApplicable(responseObserver);
+      }
+
+      @Override
+      public void readRows(
+          ReadRowsRequest request, StreamObserver<ReadRowsResponse> responseObserver) {
+        rpcBase.readRows(request, responseObserver);
+        sleepAndAbortIfApplicable(responseObserver);
+      }
+
+      @Override
+      public void mutateRows(
+          MutateRowsRequest request, StreamObserver<MutateRowsResponse> responseObserver) {
+        rpcBase.mutateRows(request, responseObserver);
+        sleepAndAbortIfApplicable(responseObserver);
+      }
+
+      @Override
+      public void readModifyWriteRow(
+          ReadModifyWriteRowRequest request,
+          StreamObserver<ReadModifyWriteRowResponse> responseObserver) {
+        rpcBase.readModifyWriteRow(request, responseObserver);
+        sleepAndAbortIfApplicable(responseObserver);
+      }
+
+      private void sleepAndAbortIfApplicable(StreamObserver<?> responseObserver) {
+        if (serverRpcAbortsForTest) {
+          try {
+            Thread.sleep(attemptTimeoutMs);
+          } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+          }
+          responseObserver.onError(new StatusRuntimeException(Status.ABORTED));
+        }
+      }
+    };
+  }
+}
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorCheckAndMutateRow.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorCheckAndMutateRow.java
new file mode 100644
index 000000000..aa630d900
--- /dev/null
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorCheckAndMutateRow.java
@@ -0,0 +1,158 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.hbase;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.CheckAndMutateRowRequest;
+import com.google.bigtable.v2.CheckAndMutateRowResponse;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.stub.StreamObserver;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang.time.StopWatch;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.Matchers;
+import org.junit.Assume;
+
+public class TestRpcRetryBehaviorCheckAndMutateRow extends TestRpcRetryBehavior {
+  private final AtomicInteger numCheckAndMutateRowInvocations = new AtomicInteger();
+
+  @Override
+  protected AtomicInteger getInvocations() {
+    // CheckAndMutateRow is not idempotent and does not support retries. Hence, we only expect 1
+    // attempt...
+    //    assertThat(numCheckAndMutateRowInvocations.get(), equalTo(1));
+    // ... Hence, avoid the regular validation method.
+    //    validateInvocations(numCheckAndMutateRowInvocations);
+    return numCheckAndMutateRowInvocations;
+  }
+
+  @Override
+  protected void executeLogic(Table table, StopWatch sw) throws Exception {
+    try {
+      RowMutations rowMutations = new RowMutations("mykey".getBytes());
+      rowMutations.add(
+          new Put(("mykey").getBytes())
+              .addColumn("cf1".getBytes(), "qualifier".getBytes(), "value".getBytes()));
+      rowMutations.add(
+          new Put(("mykey").getBytes())
+              .addColumn("cf1".getBytes(), "qualifier".getBytes(), "value".getBytes()));
+
+      sw.start();
+      table.checkAndMutate(
+          "mykey".getBytes(),
+          "b".getBytes(),
+          "c".getBytes(),
+          CompareFilter.CompareOp.EQUAL,
+          "d".getBytes(),
+          rowMutations);
+
+      fail("Should have errored out");
+    } catch (Exception e) {
+      // Stop ASAP to reduce potential flakiness (due to adding ms to measured query times).
+      sw.stop();
+
+      String expectedExceptionMessage = serverRpcAbortsForTest ? "ABORTED" : "DEADLINE_EXCEEDED";
+      assertThat(e.getCause().getMessage(), CoreMatchers.containsString(expectedExceptionMessage));
+    }
+  }
+
+  @Override
+  protected ImmutableMap.Builder<String, String> defineProperties() {
+    ImmutableMap.Builder<String, String> connPropsBuilder =
+        ImmutableMap.<String, String>builder()
+            .put(BigtableOptionsFactory.BIGTABLE_USE_TIMEOUTS_KEY, String.valueOf(timeoutEnabled))
+            .put(
+                BigtableOptionsFactory.BIGTABLE_RPC_TIMEOUT_MS_KEY,
+                String.valueOf(operationTimeoutMs))
+            .put(
+                BigtableOptionsFactory.BIGTABLE_MUTATE_RPC_TIMEOUT_MS_KEY,
+                String.valueOf(operationTimeoutMs))
+            .put(
+                BigtableOptionsFactory.BIGTABLE_LONG_RPC_TIMEOUT_MS_KEY,
+                String.valueOf(operationTimeoutMs));
+
+    if (attemptTimeoutEnabled) {
+      // Note - the property for CheckAndMutate is not yet defined; only will be if we support
+      // retries.
+      connPropsBuilder.put(
+          BigtableOptionsFactory.BIGTABLE_RPC_ATTEMPT_TIMEOUT_MS_KEY,
+          String.valueOf(attemptTimeoutMs));
+
+      // Set this to a low number to validate that it is not used when timeouts are enabled.
+      connPropsBuilder.put(BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY, "123");
+    } else {
+      connPropsBuilder.put(
+          BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY,
+          String.valueOf(maxElapsedBackoffMs));
+    }
+    return connPropsBuilder;
+  }
+
+  @Override
+  public void testOperationAndAttemptTimeoutsWillRetryOnServerAbort() throws Exception {
+    Assume.assumeTrue(
+        "Attempt RPC timeout not yet enabled for CheckAndMutate, so skipping test", false);
+    super.testOperationAndAttemptTimeoutsWillRetryOnServerAbort();
+  }
+
+  @Override
+  public void testOperationAndAttemptTimeoutsWillRetryOnServerHang() throws Exception {
+    Assume.assumeTrue(
+        "Attempt RPC timeout not yet enabled for CheckAndMutate, so skipping test", false);
+    super.testOperationAndAttemptTimeoutsWillRetryOnServerHang();
+  }
+
+  @Override
+  protected void validateOperationRuntime(StopWatch sw) {
+    if (serverRpcAbortsForTest) {
+      // CheckAndMutate currently does not retry. So if the server aborts, the RPC errors out
+      // sooner per the attempt timeout in our test, not the longer operation timeout.
+      assertThat(
+          sw.getTime(),
+          Matchers.both(Matchers.greaterThan(attemptTimeoutMs))
+              .and(Matchers.lessThan(attemptTimeoutMs + 300))); // plus some buffer
+    } else {
+      super.validateOperationRuntime(sw);
+    }
+  }
+
+  @Override
+  protected void validateInvocations(AtomicInteger counter) {
+    // CheckAndMutateRow is not idempotent and does not support retries. Hence, we only expect 1
+    // attempt...
+    assertThat(numCheckAndMutateRowInvocations.get(), equalTo(1));
+  }
+
+  @Override
+  protected BigtableGrpc.BigtableImplBase setupRpcCall() {
+    return new BigtableGrpc.BigtableImplBase() {
+      @Override
+      public void checkAndMutateRow(
+          CheckAndMutateRowRequest request,
+          StreamObserver<CheckAndMutateRowResponse> responseObserver) {
+        numCheckAndMutateRowInvocations.incrementAndGet();
+      }
+    };
+  }
+}
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorGetMultiple.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorGetMultiple.java
new file mode 100644
index 000000000..b8bc2d13e
--- /dev/null
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorGetMultiple.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.hbase;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.bigtable.v2.ReadRowsResponse;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.stub.StreamObserver;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang.time.StopWatch;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
+import org.apache.hadoop.hbase.client.Table;
+
+public class TestRpcRetryBehaviorGetMultiple extends TestRpcRetryBehavior {
+  private final AtomicInteger numReadRowsInvocations = new AtomicInteger();
+
+  @Override
+  protected AtomicInteger getInvocations() {
+    return numReadRowsInvocations;
+  }
+
+  @Override
+  protected void executeLogic(Table table, StopWatch sw) throws Exception {
+    try {
+      sw.start();
+      table.get(
+          Arrays.asList(
+              new Get("mykey1".getBytes()),
+              new Get("mykey2".getBytes()),
+              new Get("mykey3".getBytes())));
+
+      fail("Should have errored out");
+    } catch (RetriesExhaustedWithDetailsException e) {
+      // Stop ASAP to reduce potential flakiness (due to adding ms to measured query times).
+      sw.stop();
+
+      // Expect this specific error class above.
+      // As a sanity check for our test, verify that the exception count matches the number of
+      // gets (this exact
+      // verification is unnecessary for the retry functionality).
+      assertThat(e.getNumExceptions(), equalTo(3));
+    }
+  }
+
+  @Override
+  protected ImmutableMap.Builder<String, String> defineProperties() {
+    ImmutableMap.Builder<String, String> connPropsBuilder =
+        ImmutableMap.<String, String>builder()
+            .put(BigtableOptionsFactory.BIGTABLE_USE_TIMEOUTS_KEY, String.valueOf(timeoutEnabled))
+            .put(
+                BigtableOptionsFactory.BIGTABLE_READ_RPC_TIMEOUT_MS_KEY,
+                String.valueOf(operationTimeoutMs));
+
+    if (attemptTimeoutEnabled) {
+      connPropsBuilder.put(
+          BigtableOptionsFactory.BIGTABLE_READ_RPC_ATTEMPT_TIMEOUT_MS_KEY,
+          String.valueOf(attemptTimeoutMs));
+
+      // Set this to a low number to validate that it is not used when timeouts are enabled.
+      connPropsBuilder.put(BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY, "123");
+    } else {
+      connPropsBuilder.put(
+          BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY,
+          String.valueOf(maxElapsedBackoffMs));
+    }
+    return connPropsBuilder;
+  }
+
+  @Override
+  protected BigtableGrpc.BigtableImplBase setupRpcCall() {
+    return new BigtableGrpc.BigtableImplBase() {
+      @Override
+      public void readRows(
+          ReadRowsRequest request, StreamObserver<ReadRowsResponse> responseObserver) {
+        numReadRowsInvocations.incrementAndGet();
+      }
+    };
+  }
+}
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorGetSingle.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorGetSingle.java
new file mode 100644
index 000000000..6f692ef61
--- /dev/null
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorGetSingle.java
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.hbase;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.bigtable.v2.ReadRowsResponse;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.stub.StreamObserver;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang.time.StopWatch;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Table;
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.Matcher;
+import org.hamcrest.Matchers;
+
+public class TestRpcRetryBehaviorGetSingle extends TestRpcRetryBehavior {
+  private final AtomicInteger numReadRowsInvocations = new AtomicInteger();
+
+  @Override
+  protected ImmutableMap.Builder<String, String> defineProperties() {
+    ImmutableMap.Builder<String, String> connPropsBuilder =
+        ImmutableMap.<String, String>builder()
+            .put(BigtableOptionsFactory.BIGTABLE_USE_TIMEOUTS_KEY, String.valueOf(timeoutEnabled))
+            .put(
+                BigtableOptionsFactory.BIGTABLE_RPC_TIMEOUT_MS_KEY,
+                String.valueOf(operationTimeoutMs));
+
+    if (attemptTimeoutEnabled) {
+      connPropsBuilder.put(
+          BigtableOptionsFactory.BIGTABLE_RPC_ATTEMPT_TIMEOUT_MS_KEY,
+          String.valueOf(attemptTimeoutMs));
+
+      // Set this to a low number to validate that it is not used when timeouts are enabled.
+      connPropsBuilder.put(BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY, "123");
+    } else {
+      connPropsBuilder.put(
+          BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY,
+          String.valueOf(maxElapsedBackoffMs));
+    }
+    return connPropsBuilder;
+  }
+
+  @Override
+  protected void executeLogic(Table table, StopWatch sw) throws Exception {
+    try {
+      sw.start();
+      table.get(new Get("mykey".getBytes()));
+
+      fail("Should have errored out");
+    } catch (Exception e) {
+      // Stop ASAP to reduce potential flakiness (due to adding ms to measured query times).
+      sw.stop();
+
+      Matcher<String> hasAbortedMessage = CoreMatchers.containsString("ABORTED");
+      Matcher<String> hasDeadlineExceededMessage = CoreMatchers.containsString("DEADLINE_EXCEEDED");
+      if (timeoutEnabled) {
+        assertThat(e.getMessage(), hasDeadlineExceededMessage);
+      } else {
+        assertThat(
+            e.getMessage(), Matchers.either(hasDeadlineExceededMessage).or(hasAbortedMessage));
+      }
+    }
+  }
+
+  @Override
+  protected AtomicInteger getInvocations() {
+    return numReadRowsInvocations;
+  }
+
+  @Override
+  protected BigtableGrpc.BigtableImplBase setupRpcCall() {
+    return new BigtableGrpc.BigtableImplBase() {
+      @Override
+      public void readRows(
+          ReadRowsRequest request, StreamObserver<ReadRowsResponse> responseObserver) {
+        numReadRowsInvocations.incrementAndGet();
+      }
+    };
+  }
+}
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorPutMultiple.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorPutMultiple.java
new file mode 100644
index 000000000..c4b17bbb2
--- /dev/null
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorPutMultiple.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.hbase;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.MutateRowsRequest;
+import com.google.bigtable.v2.MutateRowsResponse;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.stub.StreamObserver;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang.time.StopWatch;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
+import org.apache.hadoop.hbase.client.Table;
+
+public class TestRpcRetryBehaviorPutMultiple extends TestRpcRetryBehavior {
+  private final AtomicInteger numMutateRowsInvocations = new AtomicInteger();
+
+  @Override
+  protected AtomicInteger getInvocations() {
+    return numMutateRowsInvocations;
+  }
+
+  @Override
+  protected void executeLogic(Table table, StopWatch sw) throws Exception {
+    try {
+      sw.start();
+      table.put(
+          Arrays.asList(
+              new Put(("mykey1").getBytes())
+                  .addColumn("cf1".getBytes(), "qualifier".getBytes(), "value".getBytes()),
+              new Put(("mykey2").getBytes())
+                  .addColumn("cf1".getBytes(), "qualifier".getBytes(), "value".getBytes()),
+              new Put(("mykey3").getBytes())
+                  .addColumn("cf1".getBytes(), "qualifier".getBytes(), "value".getBytes())));
+
+      fail("Should have errored out");
+    } catch (RetriesExhaustedWithDetailsException e) {
+      // Stop ASAP to reduce potential flakiness (due to adding ms to measured query times).
+      sw.stop();
+
+      // Expect this specific error class above.
+      // As a sanity check for our test, verify that the exception count matches the number of
+      // puts (this exact
+      // verification is unnecessary for the retry functionality).
+      assertThat(e.getNumExceptions(), equalTo(3));
+    }
+  }
+
+  @Override
+  protected ImmutableMap.Builder<String, String> defineProperties() {
+    ImmutableMap.Builder<String, String> connPropsBuilder =
+        ImmutableMap.<String, String>builder()
+            .put(BigtableOptionsFactory.BIGTABLE_USE_TIMEOUTS_KEY, String.valueOf(timeoutEnabled))
+            .put(
+                BigtableOptionsFactory.BIGTABLE_MUTATE_RPC_TIMEOUT_MS_KEY,
+                String.valueOf(operationTimeoutMs));
+
+    if (attemptTimeoutEnabled) {
+      // The regular RPC attempt timeout covers a singular MutateRow RPC.
+      // Use the mutate timeout instead.
+      connPropsBuilder.put(
+          BigtableOptionsFactory.BIGTABLE_MUTATE_RPC_ATTEMPT_TIMEOUT_MS_KEY,
+          String.valueOf(attemptTimeoutMs));
+
+      // Set this to a low number to validate that it is not used when timeouts are enabled.
+      connPropsBuilder.put(BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY, "123");
+    } else {
+      connPropsBuilder.put(
+          BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY,
+          String.valueOf(maxElapsedBackoffMs));
+    }
+    return connPropsBuilder;
+  }
+
+  @Override
+  protected BigtableGrpc.BigtableImplBase setupRpcCall() {
+    return new BigtableGrpc.BigtableImplBase() {
+      @Override
+      public void mutateRows(
+          MutateRowsRequest request, StreamObserver<MutateRowsResponse> responseObserver) {
+        numMutateRowsInvocations.incrementAndGet();
+      }
+    };
+  }
+}
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorPutSingle.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorPutSingle.java
new file mode 100644
index 000000000..5bbb045f9
--- /dev/null
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestRpcRetryBehaviorPutSingle.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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 com.google.cloud.bigtable.hbase;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.MutateRowRequest;
+import com.google.bigtable.v2.MutateRowResponse;
+import com.google.common.collect.ImmutableMap;
+import io.grpc.stub.StreamObserver;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang.time.StopWatch;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Table;
+import org.hamcrest.CoreMatchers;
+
+public class TestRpcRetryBehaviorPutSingle extends TestRpcRetryBehavior {
+  private final AtomicInteger numMutateRowInvocations = new AtomicInteger();
+
+  @Override
+  protected void executeLogic(Table table, StopWatch sw) throws Exception {
+    try {
+      sw.start();
+      table.put(
+          new Put(("mykey").getBytes())
+              .addColumn("cf1".getBytes(), "qualifier".getBytes(), "value".getBytes()));
+
+      fail("Should have errored out");
+    } catch (Exception e) {
+      // Stop ASAP to reduce potential flakiness (due to adding ms to measured query times).
+      sw.stop();
+
+      String expectedExceptionMessage =
+          serverRpcAbortsForTest && !timeoutEnabled ? "ABORTED" : "DEADLINE_EXCEEDED";
+      assertThat(e.getCause().getMessage(), CoreMatchers.containsString(expectedExceptionMessage));
+    }
+  }
+
+  @Override
+  protected ImmutableMap.Builder<String, String> defineProperties() {
+    ImmutableMap.Builder<String, String> connPropsBuilder =
+        ImmutableMap.<String, String>builder()
+            .put(BigtableOptionsFactory.BIGTABLE_USE_TIMEOUTS_KEY, String.valueOf(timeoutEnabled))
+            .put(
+                BigtableOptionsFactory.BIGTABLE_RPC_TIMEOUT_MS_KEY,
+                String.valueOf(operationTimeoutMs));
+
+    if (attemptTimeoutEnabled) {
+      // The regular RPC attempt timeout covers a singular MutateRow RPC.
+      connPropsBuilder.put(
+          BigtableOptionsFactory.BIGTABLE_RPC_ATTEMPT_TIMEOUT_MS_KEY,
+          String.valueOf(attemptTimeoutMs));
+
+      // Set this to a low number to validate that it is not used when timeouts are enabled.
+      connPropsBuilder.put(BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY, "123");
+    } else {
+      connPropsBuilder.put(
+          BigtableOptionsFactory.MAX_ELAPSED_BACKOFF_MILLIS_KEY,
+          String.valueOf(maxElapsedBackoffMs));
+    }
+    return connPropsBuilder;
+  }
+
+  @Override
+  protected AtomicInteger getInvocations() {
+    return numMutateRowInvocations;
+  }
+
+  @Override
+  protected BigtableGrpc.BigtableImplBase setupRpcCall() {
+    return new BigtableGrpc.BigtableImplBase() {
+      @Override
+      public void mutateRow(
+          MutateRowRequest request, StreamObserver<MutateRowResponse> responseObserver) {
+        numMutateRowInvocations.incrementAndGet();
+      }
+    };
+  }
+}
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
index 243857314..382ea1d2e 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-2.x-integration-tests</artifactId>
diff --git a/bigtable-test/bigtable-emulator-maven-plugin/pom.xml b/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
index 16a00b9f7..1db9c8f15 100644
--- a/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
+++ b/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-test</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-test:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-test:current} -->
   </parent>
 
 
diff --git a/bigtable-test/pom.xml b/bigtable-test/pom.xml
index 1248d3a51..2ff934a58 100644
--- a/bigtable-test/pom.xml
+++ b/bigtable-test/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-client-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.17.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+    <version>1.17.2-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   </parent>
 
   <artifactId>bigtable-test</artifactId>