Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1892 (2018-07-20 14:39:06)

Using gax's ExponentialRetryAlgorithm for backoffs

sduskis: @igorbernstein2  I added a bunch of comments.  I don't think that I'll have great answers for you relating to proper usage of the retry algorithm.  My main goal is to achieve a similar behavior compared to the use of ExponentialBackoff.
sduskis: @igorbernstein2, I changed the backoff algorithm slightly.  The default is to not retry if the `nextRetryDelay + timePassed > timeout`.  I updated the algorithm to wait `timeout - timePassed` in that case. 

We need some sort of boundary condition in the unit test to make sure that the sleep occurred within some boundaries so that we can determine correctness.
igorbernstein2: I’m wondering if the logic should be changed in gax instead of overlaying it here
sduskis: @igorbernstein2, we need to move forward with some sort of logic in order to fix #1858.  It would be fine to fix gax, or fix our settings later.

Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/pom.xml b/bigtable-client-core-parent/bigtable-client-core/pom.xml
index 992ee3b8d..ac4c5c702 100644
--- a/bigtable-client-core-parent/bigtable-client-core/pom.xml
+++ b/bigtable-client-core-parent/bigtable-client-core/pom.xml
@@ -92,6 +92,7 @@ limitations under the License.
         <dependency>
             <groupId>com.google.auth</groupId>
             <artifactId>google-auth-library-oauth2-http</artifactId>
+            <version>${google-auth.version}</version>
             <exclusions>
                 <exclusion>
                     <groupId>com.google.guava</groupId>
@@ -153,6 +154,17 @@ limitations under the License.
             <artifactId>grpc-netty-shaded</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.api</groupId>
+            <artifactId>gax-grpc</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.grpc</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <!-- Metrics -->
         <dependency>
             <groupId>io.opencensus</groupId>
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java
index ed311c5d8..c2edc010b 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/RetryOptions.java
@@ -67,9 +67,9 @@ public class RetryOptions implements Serializable {
    */
   public static final int DEFAULT_INITIAL_BACKOFF_MILLIS = 5;
   /**
-   * Multiplier to apply to wait times after failed retries (default value: 2).
+   * Multiplier to apply to wait times after failed retries (default value: 1.5).
    */
-  public static final double DEFAULT_BACKOFF_MULTIPLIER = 2;
+  public static final double DEFAULT_BACKOFF_MULTIPLIER = 1.5;
   /**
    * Maximum amount of time to retry before failing the operation (default value: 60 seconds).
    */
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 57ac31cc7..c7835264f 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
@@ -17,6 +17,8 @@ package com.google.cloud.bigtable.grpc;
 
 import static com.google.cloud.bigtable.grpc.io.GoogleCloudResourcePrefixInterceptor.GRPC_RESOURCE_PREFIX_KEY;
 
+import com.google.api.core.ApiClock;
+import com.google.api.core.NanoClock;
 import io.grpc.CallOptions;
 import io.grpc.Channel;
 import io.grpc.Metadata;
@@ -80,6 +82,8 @@ import com.google.common.util.concurrent.ListenableFuture;
  */
 public class BigtableDataGrpcClient implements BigtableDataClient {
 
+  private final static ApiClock CLOCK = NanoClock.getDefaultClock();
+
   // Retryable Predicates
   /** Constant <code>IS_RETRYABLE_MUTATION</code> */
   @VisibleForTesting
@@ -265,8 +269,8 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
     }
     CallOptions callOptions = getCallOptions(mutateRowsRpc.getMethodDescriptor(), request);
     Metadata metadata = createMetadata(request.getTableName());
-    return new RetryingMutateRowsOperation(
-        retryOptions, request, mutateRowsRpc, callOptions, retryExecutorService, metadata);
+    return new RetryingMutateRowsOperation(retryOptions, request, mutateRowsRpc, callOptions,
+        retryExecutorService, metadata, CLOCK);
   }
 
   /** {@inheritDoc} */
@@ -377,7 +381,7 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
     CallOptions callOptions = getCallOptions(rpc.getMethodDescriptor(), request);
     Metadata metadata = createMetadata(tableName);
     return new RetryingUnaryOperation<>(
-        retryOptions, request, rpc, callOptions, retryExecutorService, metadata);
+        retryOptions, request, rpc, callOptions, retryExecutorService, metadata, CLOCK);
   }
 
   private <ReqT, RespT> RetryingStreamOperation<ReqT, RespT> createStreamingListener(
@@ -385,7 +389,7 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
     CallOptions callOptions = getCallOptions(rpc.getMethodDescriptor(), request);
     Metadata metadata = createMetadata(tableName);
     return new RetryingStreamOperation<>(
-        retryOptions, request, rpc, callOptions, retryExecutorService, metadata);
+        retryOptions, request, rpc, callOptions, retryExecutorService, metadata, CLOCK);
   }
 
   private <ReqT> CallOptions getCallOptions(final MethodDescriptor<ReqT, ?> methodDescriptor,
@@ -497,7 +501,8 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
         readRowsAsync,
         getCallOptions(readRowsAsync.getMethodDescriptor(), request),
         retryExecutorService,
-        createMetadata(request.getTableName()));
+        createMetadata(request.getTableName()),
+        CLOCK);
   }
 
   private boolean shouldOverrideAppProfile(String requestProfile) {
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 7e5c5214e..0a8e9211d 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
@@ -17,6 +17,7 @@ package com.google.cloud.bigtable.grpc;
 
 import static com.google.cloud.bigtable.grpc.io.GoogleCloudResourcePrefixInterceptor.GRPC_RESOURCE_PREFIX_KEY;
 
+import com.google.api.core.NanoClock;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.GetSnapshotRequest;
@@ -278,8 +279,8 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
       ReqT request, BigtableAsyncRpc<ReqT, RespT> rpc, String resource) {
     CallOptions callOptions = CallOptions.DEFAULT;
     Metadata metadata = createMetadata(resource);
-    return new RetryingUnaryOperation<>(
-        retryOptions, request, rpc, callOptions, retryExecutorService, metadata);
+    return new RetryingUnaryOperation<>(retryOptions, request, rpc, callOptions,
+        retryExecutorService, metadata, NanoClock.getDefaultClock());
   }
 
   /**
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 82f51dec6..448b1a0cd 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
@@ -16,7 +16,6 @@
 package com.google.cloud.bigtable.grpc.async;
 
 import java.io.Closeable;
-import java.io.IOException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
@@ -24,9 +23,10 @@ import java.util.concurrent.TimeUnit;
 
 import javax.annotation.Nullable;
 
-import com.google.api.client.util.BackOff;
-import com.google.api.client.util.ExponentialBackOff;
-import com.google.api.client.util.Sleeper;
+import com.google.api.core.ApiClock;
+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;
@@ -52,6 +52,8 @@ import io.opencensus.trace.EndSpanOptions;
 import io.opencensus.trace.Span;
 import io.opencensus.trace.Tracer;
 import io.opencensus.trace.Tracing;
+import org.threeten.bp.Duration;
+import org.threeten.bp.temporal.ChronoUnit;
 
 /**
  * A {@link ClientCall.Listener} that retries a {@link BigtableAsyncRpc} request.
@@ -119,7 +121,9 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
     }
   }
 
-  protected BackOff currentBackoff;
+  private final ExponentialRetryAlgorithm exponentialRetryAlgorithm;
+  private final ApiClock clock;
+  private TimedAttemptSettings currentBackoff;
 
   protected final BigtableAsyncRpc<RequestT, ResponseT> rpc;
   protected final RetryOptions retryOptions;
@@ -148,6 +152,7 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
    * @param callOptions a {@link io.grpc.CallOptions} object.
    * @param retryExecutorService a {@link java.util.concurrent.ScheduledExecutorService} object.
    * @param originalMetadata a {@link io.grpc.Metadata} object.
+   * @param clock a {@link ApiClock} object
    */
   public AbstractRetryingOperation(
           RetryOptions retryOptions,
@@ -155,7 +160,8 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
           BigtableAsyncRpc<RequestT, ResponseT> retryableRpc,
           CallOptions callOptions,
           ScheduledExecutorService retryExecutorService,
-          Metadata originalMetadata) {
+          Metadata originalMetadata,
+          ApiClock clock) {
     this.retryOptions = retryOptions;
     this.request = request;
     this.rpc = retryableRpc;
@@ -164,7 +170,9 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
     this.originalMetadata = originalMetadata;
     this.completionFuture = new GrpcFuture<>();
     String spanName = makeSpanName("Operation", rpc.getMethodDescriptor().getFullMethodName());
-    operationSpan = TRACER.spanBuilder(spanName).setRecordEvents(true).startSpan();
+    this.operationSpan = TRACER.spanBuilder(spanName).setRecordEvents(true).startSpan();
+    this.clock = clock;
+    this.exponentialRetryAlgorithm = createRetryAlgorithm(clock);
   }
 
   /** {@inheritDoc} */
@@ -221,11 +229,11 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
     }
 
     // Attempt retry with backoff
-    long nextBackOff = getNextBackoff();
+    Long nextBackOff = getNextBackoff();
     failedCount += 1;
 
     // Backoffs timed out.
-    if (nextBackOff == BackOff.STOP) {
+    if (nextBackOff == null) {
       LOG.info("All retries were exhausted. Failure #%d, got: %s on channel %s.\nTrailers: %s",
           status.getCause(), failedCount, status, channelId, trailers);
       setException(getExhaustedRetriesException(status));
@@ -276,29 +284,84 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
    */
   protected abstract boolean onOK(Metadata trailers);
 
-  protected long getNextBackoff() {
+  protected Long getNextBackoff() {
     if (currentBackoff == null) {
-      currentBackoff = createBackoff();
+      // Historically, the client waited for "total timeout" after the first failure.  For now,
+      // that behavior is preserved, even though that's not the ideal.
+      //
+      // TODO: Think through retries, and create policy that works with the mental model most
+      //       users would have of relating to retries.  That would likely involve updating some
+      //       default settings in addition to changing the algorithm.
+      currentBackoff = exponentialRetryAlgorithm.createFirstAttempt();
     }
-    try {
-      return currentBackoff.nextBackOffMillis();
-    } catch (IOException e) {
-      return BackOff.STOP;
+    currentBackoff = exponentialRetryAlgorithm.createNextAttempt(currentBackoff);
+    if (!exponentialRetryAlgorithm.shouldRetry(currentBackoff)) {
+
+      // TODO: consider creating a subclass of exponentialRetryAlgorithm to encapsulate this logic
+      long timeLeftNs =  currentBackoff.getGlobalSettings().getTotalTimeout().toNanos() -
+          (clock.nanoTime() - currentBackoff.getFirstAttemptStartTimeNanos());
+      long timeLeftMs = TimeUnit.NANOSECONDS.toMillis(timeLeftNs);
+
+      if (timeLeftMs > currentBackoff.getGlobalSettings().getInitialRetryDelay().toMillis()) {
+        // The backoff algorithm doesn't always wait until the timeout is achieved.  Wait
+        // one final time so that retries hit
+        return timeLeftMs;
+      } else {
+
+        // Finish for real.
+        return null;
+      }
+    } else {
+      return currentBackoff.getRetryDelay().toMillis();
     }
   }
 
+  @VisibleForTesting
+  public boolean inRetryMode() {
+    return currentBackoff != null;
+  }
+
+  /**
+   * Either a response was found, or a timeout event occurred. Reset the information relating to
+   * Status oriented exception handling.
+   */
+  protected void resetStatusBasedBackoff() {
+    this.currentBackoff = null;
+    this.failedCount = 0;
+  }
+
   /**
    * <p>createBackoff.</p>
    *
-   * @return a {@link com.google.api.client.util.ExponentialBackOff} object.
+   * @return a {@link ExponentialRetryAlgorithm} object.
    */
-  @VisibleForTesting
-  protected ExponentialBackOff createBackoff() {
-    return new ExponentialBackOff.Builder()
-            .setInitialIntervalMillis(retryOptions.getInitialBackoffMillis())
-            .setMaxElapsedTimeMillis(retryOptions.getMaxElapsedBackoffMillis())
-            .setMultiplier(retryOptions.getBackoffMultiplier())
-            .build();
+  private ExponentialRetryAlgorithm createRetryAlgorithm(ApiClock clock) {
+    int timeoutMs = retryOptions.getMaxElapsedBackoffMillis();
+
+    RetrySettings retrySettings = RetrySettings.newBuilder()
+
+        .setJittered(true)
+
+        // How long should the sleep be between RPC failure and the next RPC retry?
+        .setInitialRetryDelay(toDuration(retryOptions.getInitialBackoffMillis()))
+
+        // How fast should the retry delay increase?
+        .setRetryDelayMultiplier(retryOptions.getBackoffMultiplier())
+
+        // What is the maximum amount of sleep time between retries?
+        // There needs to be some sane number for max retry delay, and it's unclear what that
+        // number ought to be.  1 Minute time was chosen because some number is needed.
+        .setMaxRetryDelay( Duration.of(1, ChronoUnit.MINUTES))
+
+        // How long should we wait before giving up retries after the first failure?
+        .setTotalTimeout(toDuration(timeoutMs))
+
+        .build();
+    return new ExponentialRetryAlgorithm(retrySettings, clock);
+  }
+
+  private static Duration toDuration(long millis) {
+    return Duration.of(millis, ChronoUnit.MILLIS);
   }
 
   /**
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 93f60b9b2..5cf4ab5bc 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
@@ -20,6 +20,7 @@ import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 
 import com.google.api.client.util.BackOff;
+import com.google.api.core.ApiClock;
 import com.google.bigtable.v2.MutateRowsRequest;
 import com.google.bigtable.v2.MutateRowsResponse;
 import com.google.cloud.bigtable.config.RetryOptions;
@@ -43,8 +44,9 @@ public class RetryingMutateRowsOperation extends
 
   public RetryingMutateRowsOperation(RetryOptions retryOptions, MutateRowsRequest originalRquest,
       BigtableAsyncRpc<MutateRowsRequest, MutateRowsResponse> retryableRpc, CallOptions callOptions,
-      ScheduledExecutorService retryExecutorService, Metadata originalMetadata) {
-    super(retryOptions, originalRquest, retryableRpc, callOptions, retryExecutorService, originalMetadata);
+      ScheduledExecutorService retryExecutorService, Metadata originalMetadata, ApiClock clock) {
+    super(retryOptions, originalRquest, retryableRpc, callOptions, retryExecutorService,
+        originalMetadata, clock);
     requestManager = new MutateRowsRequestManager(retryOptions, originalRquest);
     operationSpan.addAnnotation("MutationCount", ImmutableMap.of("count",
       AttributeValue.longAttributeValue(originalRquest.getEntriesCount())));
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 93581c29e..abe441d3a 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,6 +18,7 @@ package com.google.cloud.bigtable.grpc.async;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 
+import com.google.api.core.ApiClock;
 import com.google.cloud.bigtable.config.RetryOptions;
 import com.google.common.collect.ImmutableList;
 
@@ -45,11 +46,12 @@ public class RetryingStreamOperation<RequestT, ResponseT>
    * @param callOptions a {@link io.grpc.CallOptions} object.
    * @param executorService a {@link java.util.concurrent.ScheduledExecutorService} object.
    * @param metadata a {@link io.grpc.Metadata} object.
+   * @param clock a {@link ApiClock} object
    */
   public RetryingStreamOperation(RetryOptions retryOptions, RequestT request,
       BigtableAsyncRpc<RequestT, ResponseT> retryableRpc, CallOptions callOptions,
-      ScheduledExecutorService executorService, Metadata metadata) {
-    super(retryOptions, request, retryableRpc, callOptions, executorService, metadata);
+      ScheduledExecutorService executorService, Metadata metadata, ApiClock clock) {
+    super(retryOptions, request, retryableRpc, callOptions, 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 4464d0602..8acc56e1c 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,6 +18,7 @@ package com.google.cloud.bigtable.grpc.async;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import com.google.api.core.ApiClock;
 import com.google.cloud.bigtable.config.RetryOptions;
 
 import io.grpc.CallOptions;
@@ -47,11 +48,12 @@ public class RetryingUnaryOperation<RequestT, ResponseT>
    * @param callOptions a {@link io.grpc.CallOptions} object.
    * @param executorService a {@link java.util.concurrent.ScheduledExecutorService} object.
    * @param metadata a {@link io.grpc.Metadata} object.
+   * @param clock a {@link ApiClock} object
    */
   public RetryingUnaryOperation(RetryOptions retryOptions, RequestT request,
       BigtableAsyncRpc<RequestT, ResponseT> retryableRpc, CallOptions callOptions,
-      ScheduledExecutorService executorService, Metadata metadata) {
-    super(retryOptions, request, retryableRpc, callOptions, executorService, metadata);
+      ScheduledExecutorService executorService, Metadata metadata, ApiClock clock) {
+    super(retryOptions, request, retryableRpc, callOptions, 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 f60e9a172..f3d0ee175 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
@@ -16,6 +16,7 @@
 package com.google.cloud.bigtable.grpc.scanner;
 
 import com.google.api.client.util.Preconditions;
+import com.google.api.core.ApiClock;
 import com.google.cloud.bigtable.grpc.io.Watchdog.State;
 import com.google.cloud.bigtable.grpc.io.Watchdog.StreamWaitTimeoutException;
 import java.util.concurrent.ScheduledExecutorService;
@@ -118,8 +119,10 @@ public class RetryingReadRowsOperation extends
       BigtableAsyncRpc<ReadRowsRequest, ReadRowsResponse> retryableRpc,
       CallOptions callOptions,
       ScheduledExecutorService retryExecutorService,
-      Metadata originalMetadata) {
-    super(retryOptions, request, retryableRpc, callOptions, retryExecutorService, originalMetadata);
+      Metadata originalMetadata,
+      ApiClock clock) {
+    super(retryOptions, request, retryableRpc, callOptions, retryExecutorService, originalMetadata,
+        clock);
     this.rowObserver = observer;
     this.rowMerger = new RowMerger(rowObserver);
     this.adapter = new CallToStreamObserverAdapter();
@@ -249,15 +252,6 @@ public class RetryingReadRowsOperation extends
     return true;
   }
 
-  /**
-   * Either a response was found, or a timeout event occurred. Reset the information relating to
-   * Status oriented exception handling.
-   */
-  void resetStatusBasedBackoff() {
-    this.currentBackoff = null;
-    this.failedCount = 0;
-  }
-
   /**
    * Special retry handling for watchdog timeouts, which uses its own fail counter.
    *
@@ -306,11 +300,6 @@ public class RetryingReadRowsOperation extends
     return timeoutRetryCount;
   }
 
-  @VisibleForTesting
-  BackOff getCurrentBackoff() {
-    return currentBackoff;
-  }
-
   @VisibleForTesting
   RowMerger getRowMerger() {
     return rowMerger;
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
index 9627aa54b..e336a4928 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
@@ -455,6 +455,11 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
     throw new UnsupportedOperationException("cloneSnapshotAsync");
   }
 
+  @Override
+  public void cloneTableSchema(TableName tableName, TableName tableName1, boolean b) {
+    throw new UnsupportedOperationException("cloneTableSchema"); // TODO
+  }
+
   @Override
   public void compact(TableName arg0, CompactType arg1) throws IOException, InterruptedException {
     throw new UnsupportedOperationException("compact");
@@ -486,6 +491,16 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
     throw new UnsupportedOperationException("enableTableReplication");
   }
 
+  @Override
+  public Future<Void> enableReplicationPeerAsync(String s) {
+    throw new UnsupportedOperationException("enableTableReplication");
+  }
+
+  @Override
+  public Future<Void> disableReplicationPeerAsync(String s) {
+    throw new UnsupportedOperationException("disableReplicationPeerAsync");
+  }
+
   @Override
   public byte[] execProcedureWithReturn(String arg0, String arg1, Map<String, String> arg2)
       throws IOException {
@@ -674,18 +689,29 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   @Override
   public void addReplicationPeer(String arg0, ReplicationPeerConfig arg1, boolean arg2) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("addReplicationPeer"); // TODO
+  }
+
+  @Override
+  public Future<Void> addReplicationPeerAsync(String peerId, ReplicationPeerConfig peerConfig) {
+    throw new UnsupportedOperationException("addReplicationPeerAsync"); // TODO
+  }
+
+  @Override
+  public Future<Void> addReplicationPeerAsync(String s, ReplicationPeerConfig replicationPeerConfig,
+      boolean b) {
+    throw new UnsupportedOperationException("addReplicationPeerAsync"); // TODO
   }
 
   @Override
   public void appendReplicationPeerTableCFs(String arg0, Map<TableName, List<String>> arg1)
       throws ReplicationException, IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("appendReplicationPeerTableCFs"); // TODO
   }
 
   @Override
   public CacheEvictionStats clearBlockCache(TableName arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODOv
+    throw new UnsupportedOperationException("clearBlockCache"); // TODOv
   }
 
   @Override
@@ -695,12 +721,12 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   @Override
   public void disableReplicationPeer(String arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("disableReplicationPeer"); // TODO
   }
 
   @Override
   public void enableReplicationPeer(String arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("enableReplicationPeer"); // TODO
   }
 
   @Override
@@ -710,67 +736,78 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   @Override
   public List<QuotaSettings> getQuota(QuotaFilter arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("getQuota"); // TODO
   }
 
   @Override
   public List<RegionMetrics> getRegionMetrics(ServerName arg0, TableName arg1) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("getRegionMetrics"); // TODO
   }
 
   @Override
   public ReplicationPeerConfig getReplicationPeerConfig(String arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("getReplicationPeerConfig"); // TODO
   }
 
   @Override
   public boolean isMergeEnabled() throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("isMergeEnabled"); // TODO
   }
 
   @Override
   public boolean isSplitEnabled() throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("isSplitEnabled"); // TODO
   }
 
   @Override
   public List<ReplicationPeerDescription> listReplicationPeers() throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("listReplicationPeers"); // TODO
   }
 
   @Override
   public List<ReplicationPeerDescription> listReplicationPeers(Pattern arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("listReplicationPeers"); // TODO
   }
 
   @Override
   public void majorCompactRegionServer(ServerName arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("majorCompactRegionServer"); // TODO
   }
 
   @Override
   public boolean mergeSwitch(boolean arg0, boolean arg1) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("mergeSwitch"); // TODO
   }
 
   @Override
   public void removeReplicationPeer(String arg0) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("removeReplicationPeer"); // TODO
   }
 
   @Override
   public void removeReplicationPeerTableCFs(String arg0, Map<TableName, List<String>> arg1)
       throws ReplicationException, IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("removeReplicationPeerTableCFs"); // TODO
+  }
+
+  @Override
+  public Future<Void> removeReplicationPeerAsync(String s) {
+    throw new UnsupportedOperationException("removeReplicationPeerAsync"); // TODO
   }
 
   @Override
   public boolean splitSwitch(boolean arg0, boolean arg1) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("splitSwitch"); // TODO
   }
 
   @Override
   public void updateReplicationPeerConfig(String arg0, ReplicationPeerConfig arg1) throws IOException {
-    throw new UnsupportedOperationException("splitRegionAsync"); // TODO
+    throw new UnsupportedOperationException("updateReplicationPeerConfig"); // TODO
+  }
+
+  @Override
+  public Future<Void> updateReplicationPeerConfigAsync(String s,
+      ReplicationPeerConfig replicationPeerConfig) {
+    throw new UnsupportedOperationException("updateReplicationPeerConfigAsync"); // TODO
   }
 }
\ No newline at end of file
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
index 705a6dfb6..e700f0824 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
@@ -539,6 +539,12 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
 
   // ****** TO BE IMPLEMENTED [start] ******
 
+  @Override
+  public CompletableFuture<Void> cloneTableSchema(TableName tableName, TableName tableName1,
+      boolean preserveSplits) {
+    throw new UnsupportedOperationException("cloneTableSchema"); // TODO
+  }
+
   @Override
   public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern arg0) {
     throw new UnsupportedOperationException("listTableSnapshots"); // TODO
diff --git a/pom.xml b/pom.xml
index c59df3e49..70e4a6786 100644
--- a/pom.xml
+++ b/pom.xml
@@ -63,9 +63,14 @@ limitations under the License.
         <opencensus.version>0.12.3</opencensus.version>
         <guava.version>20.0</guava.version>
 
+        <!--
+           TODO: This can be removed once google-cloud-clients's versions are updated.
+        -->
+        <google-auth.version>0.10.0</google-auth.version>
+
         <!-- hbase dependency versions -->
         <hbase.version.1>1.4.5</hbase.version.1>
-        <hbase.version.2>2.0.1</hbase.version.2>
+        <hbase.version.2>2.1.0</hbase.version.2>
         <hbase.version>${hbase.version.1}</hbase.version>
         <hadoop.version>2.7.4</hadoop.version>
 


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/OperationClock.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/OperationClock.java
index 9a79a9d7c..fb153dde8 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/OperationClock.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/OperationClock.java
@@ -17,6 +17,7 @@ package com.google.cloud.bigtable.grpc.async;
 
 import com.google.api.client.util.ExponentialBackOff;
 import com.google.api.client.util.NanoClock;
+import com.google.api.core.ApiClock;
 import com.google.cloud.bigtable.config.RetryOptions;
 import org.junit.Assert;
 import org.mockito.invocation.InvocationOnMock;
@@ -34,10 +35,10 @@ import static org.mockito.Mockito.when;
  * It can configure a mock {@link ScheduledExecutorService} to set internal time, and it can
  * validate that the amount of sleep time matches expectations.
  */
-public class OperationClock implements NanoClock {
+public class OperationClock implements ApiClock {
 
   private long timeNs;
-  private long totalSleepTime = 0;
+  private long totalSleepTimeNs = 0;
 
   public OperationClock() {
     this.timeNs = System.nanoTime();
@@ -45,7 +46,12 @@ public class OperationClock implements NanoClock {
 
   @Override
   public synchronized long nanoTime() {
-    return timeNs + totalSleepTime;
+    return timeNs + totalSleepTimeNs;
+  }
+
+  @Override
+  public long millisTime() {
+    return TimeUnit.NANOSECONDS.toMillis(nanoTime());
   }
 
   /**
@@ -60,7 +66,7 @@ public class OperationClock implements NanoClock {
         long duration = invocation.getArgumentAt(1, Long.class);
         TimeUnit timeUnit = invocation.getArgumentAt(2, TimeUnit.class);
         synchronized (OperationClock.this) {
-          totalSleepTime += timeUnit.toNanos(duration);
+          totalSleepTimeNs += timeUnit.toNanos(duration);
         }
         invocation.getArgumentAt(0, Runnable.class).run();
         return future;
@@ -71,25 +77,14 @@ public class OperationClock implements NanoClock {
         .then(runAutomatically);
   }
 
-  /**
-   * Creates a {@link ExponentialBackOff} with this as its {@link NanoClock}.
-   */
-  public ExponentialBackOff createBackoff(RetryOptions retryOptions) {
-    return new ExponentialBackOff.Builder()
-        .setNanoClock(this)
-        .setInitialIntervalMillis(retryOptions.getInitialBackoffMillis())
-        .setMaxElapsedTimeMillis(retryOptions.getMaxElapsedBackoffMillis())
-        .setMultiplier(retryOptions.getBackoffMultiplier()).build();
-  }
-
   /**
    * Checks to make sure that the expected sleep time matches the actual time slept.
    */
   public synchronized void assertTimeWithinExpectations(long expectedSleepNs) {
-    long sleptSeconds = TimeUnit.NANOSECONDS.toSeconds(totalSleepTime);
-    Assert.assertTrue(String.format("Slept only %d seconds", sleptSeconds),
-        totalSleepTime >= expectedSleepNs);
-    Assert.assertTrue(String.format("Slept more than expected (%d seconds)", sleptSeconds),
-        totalSleepTime < expectedSleepNs * 2);
+    long sleptMillis = TimeUnit.MILLISECONDS.toSeconds(totalSleepTimeNs);
+    Assert.assertTrue(String.format("Slept only %d ms", sleptMillis),
+        totalSleepTimeNs >= expectedSleepNs * .9);
+    Assert.assertTrue(String.format("Slept more than expected (%d ms)", sleptMillis),
+        totalSleepTimeNs < expectedSleepNs * 1.5);
   }
 }
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 dc840c62e..50684565e 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
@@ -28,7 +28,7 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import com.google.api.client.util.ExponentialBackOff;
+import com.google.api.core.ApiClock;
 import com.google.cloud.bigtable.config.RetryOptions;
 import io.grpc.ClientCall;
 import org.junit.Assert;
@@ -217,12 +217,7 @@ public class TestRetryingMutateRowsOperation {
 
   private RetryingMutateRowsOperation createOperation(MutateRowsRequest request) {
     return new RetryingMutateRowsOperation(RETRY_OPTIONS, request, mutateRows, CallOptions.DEFAULT,
-        executorService, new Metadata()) {
-      @Override
-      protected ExponentialBackOff createBackoff() {
-        return clock.createBackoff(retryOptions);
-      }
-    };
+        executorService, new Metadata(), clock);
   }
 
   private void checkExecutor(int count) {
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 fc709b830..e70209345 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
@@ -26,7 +26,6 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import com.google.api.client.util.ExponentialBackOff;
 import com.google.cloud.bigtable.config.RetryOptions;
 import org.junit.Assert;
 import org.junit.Before;
@@ -166,13 +165,8 @@ public class TestRetryingUnaryOperation {
   }
 
   private RetryingUnaryOperation createOperation(CallOptions options) {
-    return new RetryingUnaryOperation<ReadRowsRequest, ReadRowsResponse>(RETRY_OPTIONS,
-        ReadRowsRequest.getDefaultInstance(), readAsync, options, executorService, new Metadata()) {
-      @Override
-      protected ExponentialBackOff createBackoff() {
-        return clock.createBackoff(retryOptions);
-      }
-    };
+    return new RetryingUnaryOperation<>(RETRY_OPTIONS, ReadRowsRequest.getDefaultInstance(),
+        readAsync, options, executorService, new Metadata(), clock);
   }
 
 }
\ No newline at end of file
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 76a2bbc64..4d748f485 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
@@ -25,6 +25,7 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.api.client.util.ExponentialBackOff;
+import com.google.api.core.ApiClock;
 import com.google.cloud.bigtable.config.RetryOptions;
 import com.google.cloud.bigtable.grpc.async.OperationClock;
 import com.google.cloud.bigtable.grpc.io.Watchdog;
@@ -146,11 +147,7 @@ public class RetryingReadRowsOperationTest {
   protected RetryingReadRowsOperation createOperation(CallOptions options,
       StreamObserver<FlatRow> observer) {
     return new RetryingReadRowsOperation(observer, RETRY_OPTIONS, READ_ENTIRE_TABLE_REQUEST,
-        mockRetryableRpc, options, mockRetryExecutorService, metaData) {
-      @Override protected ExponentialBackOff createBackoff() {
-        return clock.createBackoff(retryOptions);
-      }
-    };
+        mockRetryableRpc, options, mockRetryExecutorService, metaData, clock);
   }
 
   @Test
@@ -232,12 +229,9 @@ public class RetryingReadRowsOperationTest {
       Assert.assertEquals(Status.DEADLINE_EXCEEDED.getCode(), Status.fromThrowable(e).getCode());
     }
 
-    int expectedMaxElapsedTimeMs =
-        ((ExponentialBackOff) underTest.getCurrentBackoff()).getMaxElapsedTimeMillis();
-    int actualElapsedTimeMs = RETRY_OPTIONS.getMaxElapsedBackoffMillis();
+    int expectedTimeMs = RETRY_OPTIONS.getMaxElapsedBackoffMillis();
 
-    Assert.assertEquals(expectedMaxElapsedTimeMs, actualElapsedTimeMs);
-    clock.assertTimeWithinExpectations(TimeUnit.MILLISECONDS.toNanos(actualElapsedTimeMs));
+    clock.assertTimeWithinExpectations(TimeUnit.MILLISECONDS.toNanos(expectedTimeMs));
     verify(mockFlatRowObserver, times(0)).onCompleted();
     verify(mockFlatRowObserver, times(1)).onError(any(Throwable.class));
   }
@@ -349,7 +343,7 @@ public class RetryingReadRowsOperationTest {
     ByteString key2 = ByteString.copyFrom("SomeKey2", "UTF-8");
     underTest.onMessage(buildResponse(key1));
     underTest.onClose(Status.ABORTED, new Metadata());
-    Assert.assertNotNull(underTest.getCurrentBackoff());
+    Assert.assertTrue(underTest.inRetryMode());
     expectedRetryCount++;
     checkRetryRequest(underTest, key1, 9);
 
@@ -359,7 +353,7 @@ public class RetryingReadRowsOperationTest {
       expectedRetryCount++;
     }
     checkRetryRequest(underTest, key1, 9);
-    Assert.assertNull(underTest.getCurrentBackoff());
+    Assert.assertFalse(underTest.inRetryMode());
     underTest.onMessage(buildResponse(key2));
 
     for (int i = 0; i < RETRY_OPTIONS.getMaxScanTimeoutRetries(); i++) {
@@ -367,7 +361,7 @@ public class RetryingReadRowsOperationTest {
       expectedRetryCount++;
 
       performTimeout(underTest);
-      Assert.assertNull(underTest.getCurrentBackoff());
+      Assert.assertFalse(underTest.inRetryMode());
       expectedRetryCount++;
     }