Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1565 (2017-12-18 18:45:04)

Implement waitForReplication in the java client

igorbernstein2: All feedback addressed

Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/pom.xml b/bigtable-client-core-parent/bigtable-client-core/pom.xml
index 950b61253..4805a36c8 100644
--- a/bigtable-client-core-parent/bigtable-client-core/pom.xml
+++ b/bigtable-client-core-parent/bigtable-client-core/pom.xml
@@ -56,10 +56,6 @@ limitations under the License.
                     <groupId>io.grpc</groupId>
                     <artifactId>grpc-all</artifactId>
                 </exclusion>
-                <exclusion>
-                    <groupId>com.google.api</groupId>
-                    <artifactId>api-common</artifactId>
-                </exclusion>
             </exclusions>
         </dependency>
         <dependency>
@@ -71,10 +67,6 @@ limitations under the License.
                     <groupId>io.grpc</groupId>
                     <artifactId>grpc-all</artifactId>
                 </exclusion>
-                <exclusion>
-                    <groupId>com.google.api</groupId>
-                    <artifactId>api-common</artifactId>
-                </exclusion>
             </exclusions>
         </dependency>
         <dependency>
@@ -86,10 +78,6 @@ limitations under the License.
                     <groupId>io.grpc</groupId>
                     <artifactId>grpc-all</artifactId>
                 </exclusion>
-                <exclusion>
-                    <groupId>com.google.api</groupId>
-                    <artifactId>api-common</artifactId>
-                </exclusion>
             </exclusions>
         </dependency>
         <dependency>
@@ -101,10 +89,6 @@ limitations under the License.
                     <groupId>io.grpc</groupId>
                     <artifactId>grpc-all</artifactId>
                 </exclusion>
-                <exclusion>
-                    <groupId>com.google.api</groupId>
-                    <artifactId>api-common</artifactId>
-                </exclusion>
             </exclusions>
         </dependency>
         <!-- Common -->
@@ -169,20 +153,18 @@ limitations under the License.
             </exclusions>
         </dependency>
 
-        <!-- GRPC and tcnative -->
-        <!-- 
-            These dependencies are copied to bigtable-hbase-1.x and bigtable-hbase-2.x.
-
-            The copy has to happen to ensure consistency of all io.netty and io.grpc versions
-            even if there is a different transitive dependency of grpc from other clients.
-            Specifically this should avoid problems when stackdriver tracing is used in
-            conjunction with Cloud Bigtable.
-        -->
+        <!-- GRPC -->
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
             <version>${protobuff-java.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-tcnative-boringssl-static</artifactId>
+            <version>${netty-tcnative-boringssl-static.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-protobuf</artifactId>
@@ -209,12 +191,6 @@ limitations under the License.
             <version>${grpc.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>io.netty</groupId>
-            <artifactId>netty-tcnative-boringssl-static</artifactId>
-            <version>${netty-tcnative-boringssl-static.version}</version>
-        </dependency>
-
         <!-- Metrics -->
         <dependency>
             <groupId>io.opencensus</groupId>
@@ -259,6 +235,12 @@ limitations under the License.
             <version>${hamcrest.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-testing</artifactId>
+            <version>${grpc.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
index b9535e739..83cffc788 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
@@ -58,6 +58,9 @@ public class BigtableOptions implements Serializable {
   /** Constant <code>BIGTABLE_DATA_CHANNEL_COUNT_DEFAULT=getDefaultDataChannelCount()</code> */
   public static final int BIGTABLE_DATA_CHANNEL_COUNT_DEFAULT = getDefaultDataChannelCount();
 
+  /** Constant <code>BIGTABLE_APP_PROFILE_DEFAULT=""</code>, defaults to the server default app profile */
+  public static final String BIGTABLE_APP_PROFILE_DEFAULT = "";
+
   private static final Logger LOG = new Logger(BigtableOptions.class);
 
   private static int getDefaultDataChannelCount() {
@@ -76,6 +79,7 @@ public class BigtableOptions implements Serializable {
     private String userAgent;
 
     private String instanceId;
+    private String appProfileId = BIGTABLE_APP_PROFILE_DEFAULT;
 
     // Optional configuration for hosts - useful for the Bigtable team, more than anything else.
     private String dataHost = BIGTABLE_DATA_HOST_DEFAULT;
@@ -102,6 +106,7 @@ public class BigtableOptions implements Serializable {
     private Builder(BigtableOptions original) {
       this.projectId = original.projectId;
       this.instanceId = original.instanceId;
+      this.appProfileId = original.appProfileId;
       this.userAgent = original.userAgent;
       this.dataHost = original.dataHost;
       this.tableAdminHost = original.tableAdminHost;
@@ -145,6 +150,12 @@ public class BigtableOptions implements Serializable {
       return this;
     }
 
+    public Builder setAppProfileId(String appProfileId) {
+      Preconditions.checkNotNull(appProfileId);
+      this.appProfileId = appProfileId;
+      return this;
+    }
+
     public Builder setCredentialOptions(CredentialOptions credentialOptions) {
       this.credentialOptions = credentialOptions;
       return this;
@@ -246,6 +257,7 @@ public class BigtableOptions implements Serializable {
           port,
           projectId,
           instanceId,
+          appProfileId,
           userAgent,
           usePlaintextNegotiation,
           useCachedDataPool,
@@ -263,6 +275,7 @@ public class BigtableOptions implements Serializable {
   private final int port;
   private final String projectId;
   private final String instanceId;
+  private final String appProfileId;
   private final String userAgent;
   private final int dataChannelCount;
   private final boolean usePlaintextNegotiation;
@@ -283,6 +296,7 @@ public class BigtableOptions implements Serializable {
       port = 0;
       projectId = null;
       instanceId = null;
+      appProfileId = BIGTABLE_APP_PROFILE_DEFAULT;
       userAgent = null;
       dataChannelCount = 1;
       instanceName = null;
@@ -302,6 +316,7 @@ public class BigtableOptions implements Serializable {
       int port,
       String projectId,
       String instanceId,
+      String appProfileId,
       String userAgent,
       boolean usePlaintextNegotiation,
       boolean useCachedChannel,
@@ -318,6 +333,7 @@ public class BigtableOptions implements Serializable {
     this.port = port;
     this.projectId = projectId;
     this.instanceId = instanceId;
+    this.appProfileId = appProfileId;
     this.credentialOptions = credentialOptions;
     this.userAgent = userAgent;
     this.retryOptions = retryOptions;
@@ -388,6 +404,15 @@ public class BigtableOptions implements Serializable {
     return instanceId;
   }
 
+  /**
+   * <p>Getter for the field <code>appProfileId</code>.</p>
+   *
+   * @return a {@link java.lang.String} object.
+   */
+  public String getAppProfileId() {
+    return appProfileId;
+  }
+
   /**
    * <p>Getter for the field <code>port</code>.</p>
    *
@@ -488,6 +513,7 @@ public class BigtableOptions implements Serializable {
         && Objects.equals(dataHost, other.dataHost)
         && Objects.equals(projectId, other.projectId)
         && Objects.equals(instanceId, other.instanceId)
+        && Objects.equals(appProfileId, other.appProfileId)
         && Objects.equals(userAgent, other.userAgent)
         && Objects.equals(credentialOptions, other.credentialOptions)
         && Objects.equals(retryOptions, other.retryOptions)
@@ -505,6 +531,7 @@ public class BigtableOptions implements Serializable {
         .add("instanceAdminHost", instanceAdminHost)
         .add("projectId", projectId)
         .add("instanceId", instanceId)
+        .add("appProfileId", appProfileId)
         .add("userAgent", userAgent)
         .add("credentialType", credentialOptions.getCredentialType())
         .add("port", port)
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 841898119..af5dd1fec 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
@@ -51,7 +51,6 @@ public class RetryOptions implements Serializable {
   /** Constant <code>DEFAULT_ENABLE_GRPC_RETRIES_SET</code> */
   public static final Set<Status.Code> DEFAULT_ENABLE_GRPC_RETRIES_SET = ImmutableSet.of(
       Status.Code.DEADLINE_EXCEEDED,
-      Status.Code.INTERNAL,
       Status.Code.UNAVAILABLE,
       Status.Code.ABORTED,
       Status.Code.UNAUTHENTICATED);
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 b4137d866..6941bfc66 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
@@ -153,6 +153,7 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
     };
       
   // Member variables
+  private final String clientDefaultAppProfileId;
   private final ScheduledExecutorService retryExecutorService;
   private final RetryOptions retryOptions;
 
@@ -180,6 +181,7 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
       Channel channel,
       ScheduledExecutorService retryExecutorService,
       BigtableOptions bigtableOptions) {
+    this.clientDefaultAppProfileId = bigtableOptions.getAppProfileId();
     this.retryExecutorService = retryExecutorService;
     this.retryOptions = bigtableOptions.getRetryOptions();
 
@@ -232,28 +234,43 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public MutateRowResponse mutateRow(MutateRowRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
     return createUnaryListener(request, mutateRowRpc, request.getTableName()).getBlockingResult();
   }
 
   /** {@inheritDoc} */
   @Override
   public ListenableFuture<MutateRowResponse> mutateRowAsync(MutateRowRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
     return createUnaryListener(request, mutateRowRpc, request.getTableName()).getAsyncResult();
   }
 
   /** {@inheritDoc} */
   @Override
   public List<MutateRowsResponse> mutateRows(MutateRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
     return createMutateRowsOperation(request).getBlockingResult();
   }
 
   /** {@inheritDoc} */
   @Override
   public ListenableFuture<List<MutateRowsResponse>> mutateRowsAsync(MutateRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
     return createMutateRowsOperation(request).getAsyncResult();
   }
 
   private RetryingMutateRowsOperation createMutateRowsOperation(MutateRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
     CallOptions callOptions = getCallOptions(mutateRowsRpc.getMethodDescriptor(), request);
     Metadata metadata = createMetadata(request.getTableName());
     return new RetryingMutateRowsOperation(
@@ -263,6 +280,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public CheckAndMutateRowResponse checkAndMutateRow(CheckAndMutateRowRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return createUnaryListener(request, checkAndMutateRpc, request.getTableName())
         .getBlockingResult();
   }
@@ -271,12 +292,20 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   @Override
   public ListenableFuture<CheckAndMutateRowResponse> checkAndMutateRowAsync(
       CheckAndMutateRowRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return createUnaryListener(request, checkAndMutateRpc, request.getTableName()).getAsyncResult();
   }
 
   /** {@inheritDoc} */
   @Override
   public ReadModifyWriteRowResponse readModifyWriteRow(ReadModifyWriteRowRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return createUnaryListener(request, readWriteModifyRpc, request.getTableName())
         .getBlockingResult();
   }
@@ -285,6 +314,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   @Override
   public ListenableFuture<ReadModifyWriteRowResponse> readModifyWriteRowAsync(
       ReadModifyWriteRowRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return createUnaryListener(request, readWriteModifyRpc, request.getTableName())
         .getAsyncResult();
   }
@@ -292,6 +325,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public List<SampleRowKeysResponse> sampleRowKeys(SampleRowKeysRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return createStreamingListener(request, sampleRowKeysAsync, request.getTableName())
         .getBlockingResult();
   }
@@ -300,6 +337,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   @Override
   public ListenableFuture<List<SampleRowKeysResponse>> sampleRowKeysAsync(
       SampleRowKeysRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return createStreamingListener(request, sampleRowKeysAsync, request.getTableName())
         .getAsyncResult();
   }
@@ -307,6 +348,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public ListenableFuture<List<Row>> readRowsAsync(ReadRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return Futures.transform(
         createStreamingListener(request, readRowsAsync, request.getTableName()).getAsyncResult(),
         ROW_LIST_TRANSFORMER);
@@ -315,6 +360,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public ListenableFuture<List<FlatRow>> readFlatRowsAsync(ReadRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return Futures.transform(
         createStreamingListener(request, readRowsAsync, request.getTableName()).getAsyncResult(),
         FLAT_ROW_LIST_TRANSFORMER);
@@ -323,6 +372,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public List<FlatRow> readFlatRowsList(ReadRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     return FLAT_ROW_LIST_TRANSFORMER.apply(
       createStreamingListener(request, readRowsAsync, request.getTableName()).getBlockingResult());
   }
@@ -364,6 +417,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public ResultScanner<Row> readRows(ReadRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     // Delegate all resumable operations to the scanner. It will request a non-resumable scanner
     // during operation.
     // TODO(sduskis): Figure out a way to perform operation level metrics with the
@@ -401,6 +458,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public ResultScanner<FlatRow> readFlatRows(ReadRowsRequest request) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     // Delegate all resumable operations to the scanner. It will request a non-resumable scanner
     // during operation.
     final ResponseQueueReader reader = new ResponseQueueReader();
@@ -423,6 +484,10 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
   /** {@inheritDoc} */
   @Override
   public ScanHandler readFlatRows(ReadRowsRequest request, StreamObserver<FlatRow> observer) {
+    if (shouldOverrideAppProfile(request.getAppProfileId())) {
+      request = request.toBuilder().setAppProfileId(clientDefaultAppProfileId).build();
+    }
+
     RetryingReadRowsOperation operation = createReadRowsRetryListener(request, observer);
 
     // Start the operation.
@@ -442,4 +507,8 @@ public class BigtableDataGrpcClient implements BigtableDataClient {
         retryExecutorService,
         createMetadata(request.getTableName()));
   }
+
+  private boolean shouldOverrideAppProfile(String requestProfile) {
+    return !this.clientDefaultAppProfileId.isEmpty() && requestProfile.isEmpty();
+  }
 }
\ No newline at end of file
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 53c35f285..5ca190305 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
@@ -62,7 +62,6 @@ import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.internal.DnsNameResolverProvider;
 import io.grpc.internal.GrpcUtil;
-import io.netty.util.Recycler;
 
 /**
  * <p>Encapsulates the creation of Bigtable Grpc services.</p>
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java
index 2ced00570..432a9373c 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java
@@ -15,6 +15,8 @@
  */
 package com.google.cloud.bigtable.grpc;
 
+import java.util.concurrent.TimeoutException;
+
 import com.google.bigtable.admin.v2.CreateTableRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
@@ -44,7 +46,7 @@ public interface BigtableTableAdminClient {
   /**
    * Creates a new table asynchronously. The table can be created with a full set of initial column
    * families, specified in the request.
-   * 
+   *
    * @param request a {@link CreateTableRequest} object.
    */
   ListenableFuture<Table> createTableAsync(CreateTableRequest request);
@@ -127,4 +129,21 @@ public interface BigtableTableAdminClient {
    * @return a {@link ListenableFuture} that returns {@link Empty} object.
    */
   ListenableFuture<Empty> dropRowRangeAsync(DropRowRangeRequest request);
+
+  /**
+   * Blocks until replication has caught up to the point this method was called or timeout is
+   * reached.
+   *
+   * <p>This is a private alpha release of Cloud Bigtable replication. This feature
+   * is not currently available to most Cloud Bigtable customers. This feature
+   * might be changed in backward-incompatible ways and is not recommended for
+   * production use. It is not subject to any SLA or deprecation policy.
+   *
+   * @param tableName the name of the table to wait for replication.
+   * @param timeout the maximum time to wait in seconds.
+   * @throws InterruptedException if call is interrupted while waiting to recheck if replication has
+   * caught up.
+   * @throws TimeoutException if timeout is reached.
+   */
+  void waitForReplication(BigtableTableName tableName, long timeout) throws InterruptedException, TimeoutException;
 }
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 ca6aceab0..1466b660f 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,12 +17,22 @@ package com.google.cloud.bigtable.grpc;
 
 import static com.google.cloud.bigtable.grpc.io.GoogleCloudResourcePrefixInterceptor.GRPC_RESOURCE_PREFIX_KEY;
 
+import com.google.common.primitives.Ints;
+import java.io.IOException;
+import com.google.api.client.util.BackOff;
+import com.google.api.client.util.ExponentialBackOff;
+import com.google.bigtable.admin.v2.CheckConsistencyRequest;
+import com.google.bigtable.admin.v2.GenerateConsistencyTokenRequest;
+import com.google.common.annotations.VisibleForTesting;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeoutException;
 
 import com.google.bigtable.admin.v2.BigtableTableAdminGrpc;
+import com.google.bigtable.admin.v2.CheckConsistencyResponse;
 import com.google.bigtable.admin.v2.CreateTableRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
+import com.google.bigtable.admin.v2.GenerateConsistencyTokenResponse;
 import com.google.bigtable.admin.v2.GetTableRequest;
 import com.google.bigtable.admin.v2.ListTablesRequest;
 import com.google.bigtable.admin.v2.ListTablesResponse;
@@ -57,6 +67,8 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
   private final BigtableAsyncRpc<ModifyColumnFamiliesRequest, Table> modifyColumnFamilyRpc;
   private final BigtableAsyncRpc<DeleteTableRequest, Empty> deleteTableRpc;
   private final BigtableAsyncRpc<DropRowRangeRequest, Empty> dropRowRangeRpc;
+  private final BigtableAsyncRpc<GenerateConsistencyTokenRequest, GenerateConsistencyTokenResponse> generateConsistencyTokenRpc;
+  private final BigtableAsyncRpc<CheckConsistencyRequest, CheckConsistencyResponse> checkConsistencyRpc;
 
   /**
    * <p>Constructor for BigtableTableAdminGrpcClient.</p>
@@ -83,6 +95,12 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
       Predicates.<DeleteTableRequest> alwaysFalse());
     this.dropRowRangeRpc = asyncUtilities.createAsyncRpc(BigtableTableAdminGrpc.METHOD_DROP_ROW_RANGE,
       Predicates.<DropRowRangeRequest> alwaysFalse());
+    this.generateConsistencyTokenRpc = asyncUtilities.createAsyncRpc(
+        BigtableTableAdminGrpc.METHOD_GENERATE_CONSISTENCY_TOKEN,
+        Predicates.<GenerateConsistencyTokenRequest> alwaysFalse());
+    this.checkConsistencyRpc = asyncUtilities.createAsyncRpc(
+        BigtableTableAdminGrpc.METHOD_CHECK_CONSISTENCY,
+        Predicates.<CheckConsistencyRequest> alwaysFalse());
 
     this.retryOptions = bigtableOptions.getRetryOptions();
     this.retryExecutorService = retryExecutorService;
@@ -118,7 +136,7 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
     createUnaryListener(request, createTableRpc, request.getParent()).getBlockingResult();
   }
 
-  /** {@inheritDoc} 
+  /** {@inheritDoc}
    * @return */
   @Override
   public ListenableFuture<Table> createTableAsync(CreateTableRequest request) {
@@ -132,7 +150,7 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
         .getBlockingResult();
   }
 
-  /** {@inheritDoc} 
+  /** {@inheritDoc}
    * @return */
   @Override
   public ListenableFuture<Table> modifyColumnFamilyAsync(ModifyColumnFamiliesRequest request) {
@@ -145,7 +163,7 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
     createUnaryListener(request, deleteTableRpc, request.getName()).getBlockingResult();
   }
 
-  /** {@inheritDoc} 
+  /** {@inheritDoc}
    * @return */
   @Override
   public ListenableFuture<Empty> deleteTableAsync(DeleteTableRequest request) {
@@ -158,13 +176,70 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
     createUnaryListener(request, dropRowRangeRpc, request.getName()).getBlockingResult();
   }
 
-  /** {@inheritDoc} 
+  /** {@inheritDoc}
    * @return */
   @Override
   public ListenableFuture<Empty> dropRowRangeAsync(DropRowRangeRequest request) {
     return createUnaryListener(request, dropRowRangeRpc, request.getName()).getAsyncResult();
   }
 
+
+  /** {@inheritDoc} */
+  @Override
+  public void waitForReplication(BigtableTableName tableName, long timeout) throws InterruptedException, TimeoutException {
+    // A backoff that randomizes with an interval of 10s.
+    ExponentialBackOff backOff = new ExponentialBackOff.Builder()
+        .setInitialIntervalMillis(10 * 1000)
+        .setMaxIntervalMillis(10 * 1000)
+        .setMaxElapsedTimeMillis(Ints.checkedCast(timeout * 1000))
+        .build();
+
+    waitForReplication(tableName, backOff);
+  }
+
+  @VisibleForTesting
+  void waitForReplication(BigtableTableName tableName, BackOff backOff) throws InterruptedException, TimeoutException {
+    String token = generateConsistencyToken(tableName);
+
+    while (!checkConsistency(tableName, token)) {
+      long backOffMillis;
+      try {
+        backOffMillis = backOff.nextBackOffMillis();
+      } catch (IOException e) {
+        // Should never happen, we only use ExponentialBackOff which doesn't throw.
+        throw new RuntimeException("Problem getting backoff: " + e);
+      }
+      if (backOffMillis == BackOff.STOP) {
+        throw new TimeoutException(
+            "Table " + tableName.toString() + " is not consistent after timeout.");
+
+      } else {
+        // sleep for backOffMillis milliseconds and retry operation.
+        Thread.sleep(backOffMillis);
+      }
+    }
+  }
+
+  private String generateConsistencyToken(BigtableTableName tableName) {
+    GenerateConsistencyTokenRequest request =
+        GenerateConsistencyTokenRequest.newBuilder().setName(tableName.toString()).build();
+
+    return createUnaryListener(request, generateConsistencyTokenRpc, request.getName())
+        .getBlockingResult()
+        .getConsistencyToken();
+  }
+
+  private boolean checkConsistency(BigtableTableName tableName, String token) {
+    CheckConsistencyRequest request = CheckConsistencyRequest.newBuilder()
+        .setName(tableName.toString())
+        .setConsistencyToken(token)
+        .build();
+
+    return createUnaryListener(request, checkConsistencyRpc, request.getName())
+        .getBlockingResult()
+        .getConsistent();
+  }
+
   private <ReqT, RespT> RetryingUnaryOperation<ReqT, RespT> createUnaryListener(
       ReqT request, BigtableAsyncRpc<ReqT, RespT> rpc, String resource) {
     CallOptions callOptions = CallOptions.DEFAULT;
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/MutateRowsRequestManager.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/MutateRowsRequestManager.java
new file mode 100644
index 000000000..34e138820
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/MutateRowsRequestManager.java
@@ -0,0 +1,195 @@
+/*
+ * Copyright 2017 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.async;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.google.bigtable.v2.MutateRowsRequest;
+import com.google.bigtable.v2.MutateRowsResponse;
+import com.google.bigtable.v2.MutateRowsResponse.Entry;
+import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.cloud.bigtable.grpc.BigtableDataClient;
+import com.google.rpc.Status;
+
+import io.grpc.Status.Code;
+
+/**
+ * Performs retries for {@link BigtableDataClient#mutateRows(MutateRowsRequest)} operations.
+ */
+public class MutateRowsRequestManager  {
+  private final static Status STATUS_INTERNAL =
+      Status.newBuilder().setCode(io.grpc.Status.Code.INTERNAL.value()).build();
+
+  private static Code getGrpcCode(Status status) {
+    return status == null ? null : io.grpc.Status.fromCodeValue(status.getCode()).getCode();
+  }
+
+  private static Entry createEntry(int i, Status status) {
+    return MutateRowsResponse.Entry.newBuilder().setIndex(i).setStatus(status).build();
+  }
+
+  public enum ProcessingStatus {
+    /* All responses produced OK */
+    SUCCESS,
+    /* All responses produced OK or a retryable code */
+    RETRYABLE,
+    /* Some responses a non-retryable code */
+    NOT_RETRYABLE,
+    /* The response was invalid - missing indexes and etc. */
+    INVALID;
+  }
+
+  /**
+   * The current request to send. This starts as the original request. If retries occur, this
+   * request will contain the subset of Mutations that need to be retried.
+   */
+  private volatile MutateRowsRequest currentRequest;
+
+  /**
+   * When doing retries, the retry sends a partial set of the original mutations that failed with a
+   * retryable status. This array contains a mapping of indices from the {@link #currentRequest}
+   * to {@link #originalRequest}.
+   */
+  private int[] mapToOriginalIndex;
+
+  /**
+   * This array tracks the cumulative set of results across all RPC requests.
+   */
+  private final Status[] results;
+  private final RetryOptions retryOptions;
+  private final MutateRowsRequest originalRequest;
+
+  private boolean messageIsInvalid = false;
+
+  public MutateRowsRequestManager(RetryOptions retryOptions, MutateRowsRequest originalRequest) {
+    this.currentRequest = originalRequest;
+    this.originalRequest = originalRequest;
+    this.retryOptions = retryOptions;
+    results = new Status[originalRequest.getEntriesCount()];
+
+    // This map should is a map between currentRequest and originalRquest. For now, currentRequest
+    // == originalRquest, but they could diverge if a retry occurs.
+    mapToOriginalIndex = new int[originalRequest.getEntriesCount()];
+    for (int i = 0; i < mapToOriginalIndex.length; i++) {
+      mapToOriginalIndex[i] = i;
+    }
+  }
+
+  /**
+   * Adds the content of the message to the {@link #results}.
+   */
+  public void onMessage(MutateRowsResponse message) {
+    for (Entry entry : message.getEntriesList()) {
+      int index = (int) entry.getIndex();
+
+      // Sanity check to make sure that the index returned from the server is valid.
+      if (index >= mapToOriginalIndex.length || index < 0) {
+        messageIsInvalid = true;
+        break;
+      }
+
+      // Set the result.
+      results[mapToOriginalIndex[index]] = entry.getStatus();
+    }
+  }
+
+  /**
+   * This is called when all calls to {@link #onMessage(MutateRowsResponse)} are complete.
+   *
+   * @return {@link ProcessingStatus} of the accumulated responses - success, invalid, retrable,
+   *         non-retryable.
+   */
+  public ProcessingStatus onOK() {
+    // Sanity check to make sure that every mutation received a response.
+    if (!messageIsInvalid) {
+      for (int i = 0; i < results.length; i++) {
+        if (results[i] == null) {
+          messageIsInvalid = true;
+          break;
+        }
+      }
+    }
+
+    // There was a problem in the data found in onMessage(), so fail the RPC.
+    if (messageIsInvalid) {
+      return ProcessingStatus.INVALID;
+    }
+
+    List<Integer> toRetry = new ArrayList<>();
+    ProcessingStatus processingStatus = ProcessingStatus.SUCCESS;
+
+    // Check the current state to determine the state of the results.
+    // There are three states: OK, Fail, or Partial Retry.
+    for (int i = 0; i < results.length; i++) {
+      Status status = results[i];
+      if (status.getCode() == io.grpc.Status.Code.OK.value()) {
+        continue;
+      } else if (retryOptions.isRetryable(getGrpcCode(status))) {
+        // An individual mutation failed with a retryable code, usually DEADLINE_EXCEEDED.
+        toRetry.add(i);
+        if (processingStatus == ProcessingStatus.SUCCESS) {
+          processingStatus = ProcessingStatus.RETRYABLE;
+        }
+      } else {
+        // Don't retry if even a single response is not retryable.
+        processingStatus = ProcessingStatus.NOT_RETRYABLE;
+        break;
+      }
+    }
+
+    if (!toRetry.isEmpty()) {
+      currentRequest = createRetryRequest(toRetry);
+    }
+
+    return processingStatus;
+  }
+
+  public MutateRowsRequest getRetryRequest() {
+    return currentRequest;
+  }
+
+  /**
+   * Creates a new {@link MutateRowsRequest} that's a subset of the original request that
+   * corresponds to a set of indices.
+   *
+   * @param indiciesToRetry
+   * @return the new {@link MutateRowsRequest}.
+   */
+  private MutateRowsRequest createRetryRequest(List<Integer> indiciesToRetry) {
+    MutateRowsRequest.Builder updatedRequest = MutateRowsRequest.newBuilder()
+        .setTableName(originalRequest.getTableName());
+    mapToOriginalIndex = new int[indiciesToRetry.size()];
+    for (int i = 0; i < indiciesToRetry.size(); i++) {
+      mapToOriginalIndex[i] = indiciesToRetry.get(i);
+      updatedRequest.addEntries(originalRequest.getEntries(indiciesToRetry.get(i)));
+    }
+    return updatedRequest.build();
+  }
+
+  /**
+   * @return a {@link MutateRowsResponse} built from the accumulation of all calls to
+   *         onMessage/onOK.
+   */
+  public MutateRowsResponse buildResponse() {
+    List<MutateRowsResponse.Entry> entries = new ArrayList<>();
+    for (int i = 0; i < results.length; i++) {
+      Status status = (results[i] == null) ? STATUS_INTERNAL : results[i];
+      entries.add(createEntry(i, status));
+    }
+    return MutateRowsResponse.newBuilder().addAllEntries(entries).build();
+  }
+}
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 93f634f4d..62af2464e 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
@@ -15,7 +15,6 @@
  */
 package com.google.cloud.bigtable.grpc.async;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
@@ -23,15 +22,13 @@ import java.util.concurrent.ScheduledExecutorService;
 import com.google.api.client.util.BackOff;
 import com.google.bigtable.v2.MutateRowsRequest;
 import com.google.bigtable.v2.MutateRowsResponse;
-import com.google.bigtable.v2.MutateRowsResponse.Entry;
 import com.google.cloud.bigtable.config.RetryOptions;
 import com.google.cloud.bigtable.grpc.BigtableDataClient;
+import com.google.cloud.bigtable.grpc.async.MutateRowsRequestManager.ProcessingStatus;
 import com.google.common.collect.ImmutableMap;
-import com.google.rpc.Status;
 
 import io.grpc.CallOptions;
 import io.grpc.Metadata;
-import io.grpc.Status.Code;
 import io.opencensus.trace.AttributeValue;
 
 /**
@@ -41,180 +38,60 @@ public class RetryingMutateRowsOperation extends
     AbstractRetryingOperation<MutateRowsRequest, MutateRowsResponse, List<MutateRowsResponse>> {
   private final static io.grpc.Status INVALID_RESPONSE =
       io.grpc.Status.INTERNAL.withDescription("The server returned an invalid response");
-  private final static Status STATUS_INTERNAL =
-      Status.newBuilder().setCode(io.grpc.Status.Code.INTERNAL.value()).build();
 
-  private static Code getGrpcCode(Status status) {
-    return status == null ? null : io.grpc.Status.fromCodeValue(status.getCode()).getCode();
-  }
-
-  private static Entry createEntry(int i, Status status) {
-    return MutateRowsResponse.Entry.newBuilder().setIndex(i).setStatus(status).build();
-  }
-
-
-  /**
-   * The current request to send. This starts as the original request. If retries occur, this
-   * request will contain the subset of Mutations that need to be retried.
-   */
-  private MutateRowsRequest currentRequest;
-
-  /**
-   * When doing retries, the retry sends a partial set of the original mutations that failed with a
-   * retryable status. This array contains a mapping of indices from the {@link #getRetryRequest()}
-   * to {@link #getOriginalRequest()}.
-   */
-  private int[] mapToOriginalIndex;
-
-  /**
-   * This array tracks the cumulative set of results across all RPC requests.
-   */
-  private final Status[] results;
-  private boolean messageIsInvalid;
+  private final MutateRowsRequestManager requestManager;
 
   public RetryingMutateRowsOperation(RetryOptions retryOptions, MutateRowsRequest originalRquest,
       BigtableAsyncRpc<MutateRowsRequest, MutateRowsResponse> retryableRpc, CallOptions callOptions,
       ScheduledExecutorService retryExecutorService, Metadata originalMetadata) {
     super(retryOptions, originalRquest, retryableRpc, callOptions, retryExecutorService, originalMetadata);
-    this.currentRequest = originalRquest;
-    results = new Status[originalRquest.getEntriesCount()];
-
-    // This map should is a map between currentRequest and originalRquest. For now, currentRequest
-    // == originalRquest, but they could diverge if a retry occurs.
-    mapToOriginalIndex = new int[originalRquest.getEntriesCount()];
-    for (int i = 0; i < mapToOriginalIndex.length; i++) {
-      mapToOriginalIndex[i] = i;
-    }
+    requestManager = new MutateRowsRequestManager(retryOptions, originalRquest);
     operationSpan.addAnnotation("MutationCount", ImmutableMap.of("count",
       AttributeValue.longAttributeValue(originalRquest.getEntriesCount())));
   }
 
-  /**
-   * Adds the content of the message to the {@link #results}.
-   */
   @Override
   public void onMessage(MutateRowsResponse message) {
-    for (Entry entry : message.getEntriesList()) {
-      int index = (int) entry.getIndex();
-
-      // Sanity check to make sure that the index returned from the server is valid.
-      if (index >= mapToOriginalIndex.length || index < 0) {
-        messageIsInvalid = true;
-        break;
-      }
-
-      // Set the result.
-      results[mapToOriginalIndex[index]] = entry.getStatus();
-    }
+    requestManager.onMessage(message);
   }
 
   @Override
   protected MutateRowsRequest getRetryRequest() {
-    return currentRequest;
+    return requestManager.getRetryRequest();
   }
 
   @Override
   protected boolean onOK(Metadata trailers) {
-    // Sanity check to make sure that every mutation received a response.
-    for (int i = 0; i < results.length; i++) {
-      if (results[i] == null) {
-        messageIsInvalid = true;
-        break;
-      }
-    }
+    ProcessingStatus status = requestManager.onOK();
 
-    // There was a problem in the data found in onMessage(), so fail the RPC.
-    if (messageIsInvalid) {
+    if (status == ProcessingStatus.INVALID) {
+      // Set an exception.
       onError(INVALID_RESPONSE, trailers);
-      return false;
-    }
-
-    boolean fail = false;
-    List<Integer> toRetry = new ArrayList<>();
-
-    // Check the current state to determine the state of the results.
-    // There are three states: OK, Fail, or Partial Retry.
-    for (int i = 0; i < results.length; i++) {
-      Status status = results[i];
-      if (status.getCode() == io.grpc.Status.Code.OK.value()) {
-        continue;
-      }
-
-      if (retryOptions.isRetryable(getGrpcCode(status))) {
-        // An individual mutation failed with a retryable code, usually DEADLINE_EXCEEDED.
-        toRetry.add(i);
-      } else {
-        // Don't retry if even a single response is not retryable.
-        fail = true;
-      }
-    }
-
-    if (fail) {
-      rpc.getRpcMetrics().markFailure();
-    }
-
-    if (!toRetry.isEmpty()) {
-      operationSpan.addAnnotation("MutationCount", ImmutableMap.of("retryCount",
-        AttributeValue.longAttributeValue(toRetry.size())));
+      return true;
     }
 
-    // OK or Fail should set the future, and end the operation.
-    // The caller should handle the processing of the failed mutations.
-    if (toRetry.isEmpty() || fail) {
-      completionFuture.set(Arrays.asList(buildResponse()));
+    // There was a problem in the data found in onMessage(), so fail the RPC.
+    if (status == ProcessingStatus.SUCCESS || status == ProcessingStatus.NOT_RETRYABLE) {
+      // Set the response, with either success, or non-retryable responses.
+      completionFuture.set(Arrays.asList(requestManager.buildResponse()));
       return true;
     }
 
     // Perform a partial retry, if the backoff policy allows it.
     long nextBackOff = getNextBackoff();
     if (nextBackOff == BackOff.STOP) {
+      // Return the response as is, and don't retry;
       rpc.getRpcMetrics().markRetriesExhasted();
-      completionFuture.set(Arrays.asList(buildResponse()));
+      completionFuture.set(Arrays.asList(requestManager.buildResponse()));
+      operationSpan.addAnnotation("MutationCount", ImmutableMap.of("failureCount",
+        AttributeValue.longAttributeValue(requestManager.getRetryRequest().getEntriesCount())));
       return true;
-    } else {
-      currentRequest = createRetryRequest(toRetry);
-      mapToOriginalIndex = toIntArray(toRetry);
-      performRetry(nextBackOff);
-      return false;
-    }
-  }
-
-  /**
-   * Creates a new {@link MutateRowsRequest} that's a subset of the original request that
-   * corresponds to a set of indices.
-   *
-   * @param indiciesToRetry
-   * @return the new {@link MutateRowsRequest}.
-   */
-  private MutateRowsRequest createRetryRequest(List<Integer> indiciesToRetry) {
-    MutateRowsRequest originalRequest = getOriginalRequest();
-    MutateRowsRequest.Builder updatedRequest = MutateRowsRequest.newBuilder()
-        .setTableName(originalRequest.getTableName());
-
-    for (int i = 0; i < indiciesToRetry.size(); i++) {
-      updatedRequest.addEntries(originalRequest.getEntries(indiciesToRetry.get(i)));
     }
 
-    return updatedRequest.build();
-  }
-
-  /**
-   * Converts the List<Integer> to int[].
-   */
-  private int[] toIntArray(List<Integer> list) {
-    int[] array = new int[list.size()];
-    for (int i = 0; i < list.size(); i++) {
-      array[i] = list.get(i);
-    }
-    return array;
+    performRetry(nextBackOff);
+    operationSpan.addAnnotation("MutationCount", ImmutableMap.of("retryCount",
+      AttributeValue.longAttributeValue(requestManager.getRetryRequest().getEntriesCount())));
+    return false;
   }
 
-  private MutateRowsResponse buildResponse() {
-    List<MutateRowsResponse.Entry> entries = new ArrayList<>();
-    for (int i = 0; i < results.length; i++) {
-      Status status = (results[i] == null) ? STATUS_INTERNAL : results[i];
-      entries.add(createEntry(i, status));
-    }
-    return MutateRowsResponse.newBuilder().addAllEntries(entries).build();
-  }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
index a920fc310..0f84e2cc7 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
@@ -38,9 +38,9 @@ class ReadRowsRequestManager {
   private final ReadRowsRequest originalRequest;
 
   // The number of rows read so far.
-  private long rowCount = 0;
+  private volatile long rowCount = 0;
 
-  private ByteString lastFoundKey;
+  private volatile ByteString lastFoundKey;
 
   /**
    * <p>
@@ -54,13 +54,17 @@ class ReadRowsRequestManager {
 
   void updateLastFoundKey(ByteString key) {
     this.lastFoundKey = key;
-    rowCount++;
+  }
+
+  void incrementRowCount(int count) {
+    rowCount += count;
   }
 
   ReadRowsRequest buildUpdatedRequest() {
     ReadRowsRequest.Builder newRequest = ReadRowsRequest.newBuilder()
         .setRows(filterRows())
-        .setTableName(originalRequest.getTableName());
+        .setTableName(originalRequest.getTableName())
+        .setAppProfileId(originalRequest.getAppProfileId());
 
     if (originalRequest.hasFilter()) {
       newRequest.setFilter(originalRequest.getFilter());
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 1f773d475..12c81c572 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
@@ -118,6 +118,7 @@ public class RetryingReadRowsOperation extends
   private CallToStreamObserverAdapter<ReadRowsRequest> adapter;
   private StreamObserver<ReadRowsResponse> resultObserver;
   private int totalRowsProcessed = 0;
+  private volatile ReadRowsRequest nextRequest;
 
   public RetryingReadRowsOperation(
       StreamObserver<FlatRow> observer,
@@ -130,6 +131,7 @@ public class RetryingReadRowsOperation extends
     super(retryOptions, request, retryableRpc, callOptions, retryExecutorService, originalMetadata);
     this.rowObserver = observer;
     this.requestManager = new ReadRowsRequestManager(request);
+    this.nextRequest = request;
   }
 
   // This observer will be notified after ReadRowsResponse have been fully processed.
@@ -142,7 +144,7 @@ public class RetryingReadRowsOperation extends
    */
   @Override
   protected ReadRowsRequest getRetryRequest() {
-    return requestManager.buildUpdatedRequest();
+    return nextRequest;
   }
 
   @SuppressWarnings("unchecked")
@@ -182,6 +184,7 @@ public class RetryingReadRowsOperation extends
       AttributeValue.longAttributeValue(rowCountInLastMessage)));
 
     totalRowsProcessed += rowCountInLastMessage;
+    requestManager.incrementRowCount(rowCountInLastMessage);
 
     ByteString lastProcessedKey = rowMerger.getLastCompletedRowKey();
     if (previouslyProcessedKey != lastProcessedKey) {
@@ -305,6 +308,17 @@ public class RetryingReadRowsOperation extends
     }
   }
 
+  @Override
+  protected void performRetry(long nextBackOff) {
+    buildUpdatedRequst();
+    super.performRetry(nextBackOff);
+  }
+
+  @VisibleForTesting
+  ReadRowsRequest buildUpdatedRequst() {
+    return nextRequest = requestManager.buildUpdatedRequest();
+  }
+
   @VisibleForTesting
   int getTimeoutRetryCount() {
     return timeoutRetryCount;
diff --git a/bigtable-hbase-parent/bigtable-hbase-1.x/pom.xml b/bigtable-hbase-parent/bigtable-hbase-1.x/pom.xml
index e89e95b99..cf49de64f 100644
--- a/bigtable-hbase-parent/bigtable-hbase-1.x/pom.xml
+++ b/bigtable-hbase-parent/bigtable-hbase-1.x/pom.xml
@@ -39,54 +39,6 @@ limitations under the License.
       <version>${project.version}</version>
     </dependency>
 
-    <!-- GRPC and tcnative -->
-    <!-- 
-        These dependencies are copied to bigtable-hbase-1.x and bigtable-hbase-2.x.
-
-        The copy has to happen to ensure consistency of all io.netty and io.grpc versions
-        even if there is a different transitive dependency of grpc from other clients.
-        Specifically this should avoid problems when stackdriver tracing is used in
-        conjunction with Cloud Bigtable.
-    -->
-    <dependency>
-        <groupId>com.google.protobuf</groupId>
-        <artifactId>protobuf-java</artifactId>
-        <version>${protobuff-java.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-protobuf</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-auth</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-core</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-netty</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-stub</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-
-    <dependency>
-        <groupId>io.netty</groupId>
-        <artifactId>netty-tcnative-boringssl-static</artifactId>
-        <version>${netty-tcnative-boringssl-static.version}</version>
-    </dependency>
-
-    <!-- END GRPC and tcnative -->
-
     <dependency>
       <groupId>io.dropwizard.metrics</groupId>
       <artifactId>metrics-graphite</artifactId>
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/pom.xml b/bigtable-hbase-parent/bigtable-hbase-2.x/pom.xml
index 4da7e6b2c..bb0e7b02a 100644
--- a/bigtable-hbase-parent/bigtable-hbase-2.x/pom.xml
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/pom.xml
@@ -55,54 +55,6 @@ limitations under the License.
        </exclusions>
    </dependency>
 
-    <!-- GRPC and tcnative -->
-    <!-- 
-        These dependencies are copied to bigtable-hbase-1.x and bigtable-hbase-2.x.
-
-        The copy has to happen to ensure consistency of all io.netty and io.grpc versions
-        even if there is a different transitive dependency of grpc from other clients.
-        Specifically this should avoid problems when stackdriver tracing is used in
-        conjunction with Cloud Bigtable.
-    -->
-    <dependency>
-        <groupId>com.google.protobuf</groupId>
-        <artifactId>protobuf-java</artifactId>
-        <version>${protobuff-java.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-protobuf</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-auth</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-core</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-netty</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-    <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-stub</artifactId>
-        <version>${grpc.version}</version>
-    </dependency>
-
-    <dependency>
-        <groupId>io.netty</groupId>
-        <artifactId>netty-tcnative-boringssl-static</artifactId>
-        <version>${netty-tcnative-boringssl-static.version}</version>
-    </dependency>
-
-    <!-- END GRPC and tcnative -->
-
     <dependency>
       <groupId>io.dropwizard.metrics</groupId>
       <artifactId>metrics-graphite</artifactId>
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
new file mode 100644
index 000000000..324b75694
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
@@ -0,0 +1,731 @@
+/*
+ * Copyright 2017 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.hbase2_x;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hadoop.hbase.ClusterStatus;
+import org.apache.hadoop.hbase.ClusterStatus.Option;
+import org.apache.hadoop.hbase.NamespaceDescriptor;
+import org.apache.hadoop.hbase.RegionLoad;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNotEnabledException;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.AsyncAdmin;
+import org.apache.hadoop.hbase.client.BigtableAsyncConnection;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.CompactionState;
+import org.apache.hadoop.hbase.client.RawAsyncTable.CoprocessorCallable;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.SnapshotDescription;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.replication.TableCFs;
+import org.apache.hadoop.hbase.client.security.SecurityCapability;
+import org.apache.hadoop.hbase.quotas.QuotaFilter;
+import org.apache.hadoop.hbase.quotas.QuotaSettings;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
+import org.apache.hadoop.hbase.shaded.com.google.protobuf.RpcChannel;
+import org.apache.hadoop.hbase.util.Bytes;
+import com.google.bigtable.admin.v2.CreateTableRequest;
+import com.google.bigtable.admin.v2.CreateTableRequest.Split;
+import com.google.bigtable.admin.v2.DeleteTableRequest;
+import com.google.bigtable.admin.v2.DeleteTableRequest.Builder;
+import com.google.bigtable.admin.v2.ListTablesRequest;
+import com.google.bigtable.admin.v2.Table;
+import com.google.cloud.bigtable.config.BigtableOptions;
+import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.grpc.BigtableInstanceName;
+import com.google.cloud.bigtable.grpc.BigtableTableAdminClient;
+import com.google.cloud.bigtable.hbase.adapters.admin.ColumnDescriptorAdapter;
+import com.google.cloud.bigtable.hbase2_x.adapters.admin.TableAdapter2x;
+import com.google.protobuf.ByteString;
+
+/**
+ * Bigtable implementation of {@link AsyncAdmin}
+ * 
+ * @author spollapally
+ */
+public class BigtableAsyncAdmin implements AsyncAdmin {
+  private final Logger LOG = new Logger(getClass());
+
+  private final Set<TableName> disabledTables;
+  private final BigtableOptions options;
+  private final BigtableTableAdminClient bigtableTableAdminClient;
+  private BigtableInstanceName bigtableInstanceName;
+  private final TableAdapter2x tableAdapter2x;
+
+  public BigtableAsyncAdmin(BigtableAsyncConnection asyncConnection) throws IOException {
+    LOG.debug("Creating BigtableAsyncAdmin");
+    this.options = asyncConnection.getOptions();
+    this.bigtableTableAdminClient = asyncConnection.getSession().getTableAdminClient();
+    this.disabledTables = asyncConnection.getDisabledTables();
+    this.bigtableInstanceName = options.getInstanceName();
+    this.tableAdapter2x = new TableAdapter2x(options, new ColumnDescriptorAdapter());
+  }
+
+  @Override
+  public CompletableFuture<Void> createTable(TableDescriptor desc, Optional<byte[][]> splitKeys) {
+    // wraps exceptions in a CF (CompletableFuture). No null check here on desc to match Hbase impl
+    if (desc.getTableName() == null) {
+      return FutureUtils.failedFuture(new IllegalArgumentException("TableName cannot be null"));
+    }
+
+    // Using this pattern of wrapping inexpensive prep code is required to keep all exception
+    // handing within a CF, that way clients don't have to handle them differently.
+    return CompletableFuture.supplyAsync(() -> {
+      CreateTableRequest.Builder builder = CreateTableRequest.newBuilder();
+      builder.setParent(bigtableInstanceName.toString());
+      builder.setTableId(desc.getTableName().getQualifierAsString());
+      builder.setTable(tableAdapter2x.adapt(desc));
+      if (splitKeys.isPresent()) {
+        for (byte[] splitKey : splitKeys.get()) {
+          builder
+              .addInitialSplits(Split.newBuilder().setKey(ByteString.copyFrom(splitKey)).build());
+        }
+      }
+      return builder;
+    }).thenCompose(
+        r -> FutureUtils.toCompletableFuture(bigtableTableAdminClient.createTableAsync(r.build())))
+        .thenAccept(r -> {
+        });
+  }
+
+  @Override
+  public CompletableFuture<Void> createTable(TableDescriptor desc, byte[] startKey, byte[] endKey,
+      int numRegions) {
+
+    return CompletableFuture.supplyAsync(() -> {
+      Optional<byte[][]> splitKeys = Optional.empty();
+      if (numRegions < 3) {
+        throw new IllegalArgumentException("Must create at least three regions");
+      } else if (Bytes.compareTo(startKey, endKey) >= 0) {
+        throw new IllegalArgumentException("Start key must be smaller than end key");
+      }
+      if (numRegions == 3) {
+        splitKeys = Optional.ofNullable(new byte[][] {startKey, endKey});
+      } else {
+        splitKeys = Optional.ofNullable(Bytes.split(startKey, endKey, numRegions - 3));
+        if (!splitKeys.isPresent() || splitKeys.get().length != numRegions - 1) {
+          throw new IllegalArgumentException("Unable to split key range into enough regions");
+        }
+      }
+      return splitKeys;
+    }).thenCompose(skeys -> createTable(desc, skeys));
+  }
+
+  @Override
+  public CompletableFuture<Void> disableTable(TableName tableName) {
+    CompletableFuture<Void> cf = new CompletableFuture<>();
+    tableExists(tableName).thenAccept(exists -> {
+      if (!exists) {
+        cf.completeExceptionally(new TableNotFoundException(tableName));
+        return;
+      }
+      if (disabledTables.contains(tableName)) {
+        cf.completeExceptionally(new TableNotEnabledException(tableName));
+        return;
+      }
+      disabledTables.add(tableName);
+      LOG.warn("Table " + tableName + " was disabled in memory only.");
+      cf.complete(null);
+    });
+
+    return cf;
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteTable(TableName tableName) {
+    return CompletableFuture.supplyAsync(() -> {
+      Builder deleteBuilder = DeleteTableRequest.newBuilder();
+      deleteBuilder.setName(bigtableInstanceName.toTableNameStr(tableName.getNameAsString()));
+      return deleteBuilder;
+    }).thenCompose(
+        d -> FutureUtils.toCompletableFuture(bigtableTableAdminClient.deleteTableAsync(d.build())))
+        .thenAccept(r -> {
+          disabledTables.remove(tableName);
+        });
+  }
+
+  @Override
+  public CompletableFuture<Boolean> tableExists(TableName tableName) {
+    return listTableNames(Optional.of(Pattern.compile(tableName.getNameAsString())), false)
+        .thenApply(r -> r.stream().anyMatch(e -> e.equals(tableName)));
+  }
+
+  @Override
+  public CompletableFuture<List<TableName>> listTableNames(Optional<Pattern> tableNamePattern,
+      boolean includeSysTables) {
+    return requestTableList().thenApply(r -> {
+      Stream<TableName> result;
+      if (tableNamePattern.isPresent()) {
+        result = r.stream().map(e -> bigtableInstanceName.toTableId(e.getName()))
+            .filter(e -> tableNamePattern.get().matcher(e).matches())
+            .map(e -> TableName.valueOf(e));
+      } else {
+        result = r.stream().map(e -> bigtableInstanceName.toTableId(e.getName()))
+            .map(e -> TableName.valueOf(e));
+      }
+      return result.collect(Collectors.toList());
+    });
+  }
+
+  @Override
+  public CompletableFuture<List<TableDescriptor>> listTables(Optional<Pattern> tableNamePattern,
+      boolean includeSysTables) {
+    return requestTableList().thenApply(r -> {
+      List<TableDescriptor> result = new ArrayList<>();
+      Boolean hasNonEmptyPattern = tableNamePattern.isPresent();
+      for (Table table : r) {
+        String tableName = bigtableInstanceName.toTableId(table.getName());
+        if (hasNonEmptyPattern) {
+          if (tableNamePattern.get().matcher(tableName).matches()) {
+            result.add(tableAdapter2x.adapt(table));
+          }
+        } else {
+          result.add(tableAdapter2x.adapt(table));
+        }
+      }
+      return result;
+    });
+  }
+
+  private CompletableFuture<List<Table>> requestTableList() {
+    return CompletableFuture.supplyAsync(() -> {
+      ListTablesRequest.Builder builder = ListTablesRequest.newBuilder();
+      builder.setParent(bigtableInstanceName.toString());
+      return builder;
+    }).thenCompose(
+        b -> FutureUtils.toCompletableFuture(bigtableTableAdminClient.listTablesAsync(b.build())))
+        .thenApply(r -> r.getTablesList());
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isTableDisabled(TableName tableName) {
+    return CompletableFuture.completedFuture(disabledTables.contains(tableName));
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isTableEnabled(TableName tableName) {
+    return CompletableFuture.completedFuture(!disabledTables.contains(tableName));
+  }
+
+  @Override
+  public CompletableFuture<Boolean> abortProcedure(long arg0, boolean arg1) {
+    throw new UnsupportedOperationException("abortProcedure"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> addColumnFamily(TableName arg0, ColumnFamilyDescriptor arg1) {
+    throw new UnsupportedOperationException("addColumnFamily"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> addReplicationPeer(String arg0, ReplicationPeerConfig arg1) {
+    throw new UnsupportedOperationException("addReplicationPeer"); // TODO
+
+  }
+
+  @Override
+  public CompletableFuture<Void> appendReplicationPeerTableCFs(String arg0,
+      Map<TableName, ? extends Collection<String>> arg1) {
+    throw new UnsupportedOperationException("appendReplicationPeerTableCFs"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> assign(byte[] arg0) {
+    throw new UnsupportedOperationException("assign"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> balance(boolean arg0) {
+    throw new UnsupportedOperationException("balance"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> clearCompactionQueues(ServerName arg0, Set<String> arg1) {
+    throw new UnsupportedOperationException("clearCompactionQueues"); // TODO
+
+  }
+
+  @Override
+  public CompletableFuture<List<ServerName>> clearDeadServers(List<ServerName> arg0) {
+    throw new UnsupportedOperationException("clearDeadServers"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> cloneSnapshot(String arg0, TableName arg1) {
+    throw new UnsupportedOperationException("cloneSnapshot"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> compact(TableName arg0, Optional<byte[]> arg1) {
+    throw new UnsupportedOperationException("compact"); // TODO
+
+  }
+
+  @Override
+  public CompletableFuture<Void> compactRegion(byte[] arg0, Optional<byte[]> arg1) {
+    throw new UnsupportedOperationException("compactRegion"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> compactRegionServer(ServerName arg0) {
+    throw new UnsupportedOperationException("compactRegionServer"); // TODO
+  }
+
+  @Override
+  public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> arg0,
+      CoprocessorCallable<S, R> arg1) {
+    throw new UnsupportedOperationException("coprocessorService"); // TODO
+  }
+
+  @Override
+  public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> arg0,
+      CoprocessorCallable<S, R> arg1, ServerName arg2) {
+    throw new UnsupportedOperationException("coprocessorService"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> createNamespace(NamespaceDescriptor arg0) {
+    throw new UnsupportedOperationException("createNamespace"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> decommissionRegionServers(List<ServerName> arg0, boolean arg1) {
+    throw new UnsupportedOperationException("decommissionRegionServers"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteColumnFamily(TableName arg0, byte[] arg1) {
+    throw new UnsupportedOperationException("deleteColumnFamily"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteNamespace(String arg0) {
+    throw new UnsupportedOperationException("deleteNamespace"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteSnapshot(String arg0) {
+    throw new UnsupportedOperationException("deleteSnapshot"); // TODO
+
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteTableSnapshots(Pattern arg0, Pattern arg1) {
+    throw new UnsupportedOperationException("deleteTableSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> disableReplicationPeer(String arg0) {
+    throw new UnsupportedOperationException("disableReplicationPeer"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> enableReplicationPeer(String arg0) {
+    throw new UnsupportedOperationException("enableReplicationPeer"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> enableTable(TableName arg0) {
+    throw new UnsupportedOperationException("enableTable"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> execProcedure(String arg0, String arg1, Map<String, String> arg2) {
+    throw new UnsupportedOperationException("execProcedure"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<byte[]> execProcedureWithRet(String arg0, String arg1,
+      Map<String, String> arg2) {
+    throw new UnsupportedOperationException("execProcedureWithRet"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> flush(TableName arg0) {
+    throw new UnsupportedOperationException("flush"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> flushRegion(byte[] arg0) {
+    throw new UnsupportedOperationException("flushRegion"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<ClusterStatus> getClusterStatus() {
+    throw new UnsupportedOperationException("getClusterStatus"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<ClusterStatus> getClusterStatus(EnumSet<Option> arg0) {
+    throw new UnsupportedOperationException("getClusterStatus"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<CompactionState> getCompactionState(TableName arg0) {
+    throw new UnsupportedOperationException("getCompactionState"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<CompactionState> getCompactionStateForRegion(byte[] arg0) {
+    throw new UnsupportedOperationException("getCompactionStateForRegion"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Optional<Long>> getLastMajorCompactionTimestamp(TableName arg0) {
+    throw new UnsupportedOperationException("getLastMajorCompactionTimestamp"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Optional<Long>> getLastMajorCompactionTimestampForRegion(byte[] arg0) {
+    throw new UnsupportedOperationException("getLastMajorCompactionTimestampForRegion"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<String> getLocks() {
+    throw new UnsupportedOperationException("getLocks"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<NamespaceDescriptor> getNamespaceDescriptor(String arg0) {
+    throw new UnsupportedOperationException("getNamespaceDescriptor"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<RegionInfo>> getOnlineRegions(ServerName arg0) {
+    throw new UnsupportedOperationException("getOnlineRegions"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<String> getProcedures() {
+    throw new UnsupportedOperationException("getProcedures"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<QuotaSettings>> getQuota(QuotaFilter arg0) {
+    throw new UnsupportedOperationException("getQuota"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName arg0,
+      Optional<TableName> arg1) {
+    throw new UnsupportedOperationException("getRegionLoads"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<ReplicationPeerConfig> getReplicationPeerConfig(String arg0) {
+    throw new UnsupportedOperationException("getReplicationPeerConfig"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<SecurityCapability>> getSecurityCapabilities() {
+    throw new UnsupportedOperationException("getSecurityCapabilities"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<TableDescriptor> getTableDescriptor(TableName arg0) {
+    throw new UnsupportedOperationException("getTableDescriptor"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<RegionInfo>> getTableRegions(TableName arg0) {
+    throw new UnsupportedOperationException("getTableRegions"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isBalancerOn() {
+    throw new UnsupportedOperationException("isBalancerOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isCatalogJanitorOn() {
+    throw new UnsupportedOperationException("isCatalogJanitorOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isCleanerChoreOn() {
+    throw new UnsupportedOperationException("isCleanerChoreOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isMasterInMaintenanceMode() {
+    throw new UnsupportedOperationException("isMasterInMaintenanceMode"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isMergeOn() {
+    throw new UnsupportedOperationException("isMergeOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isNormalizerOn() {
+    throw new UnsupportedOperationException("isNormalizerOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isProcedureFinished(String arg0, String arg1,
+      Map<String, String> arg2) {
+    throw new UnsupportedOperationException("isProcedureFinished"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isSnapshotFinished(SnapshotDescription arg0) {
+    throw new UnsupportedOperationException("isSnapshotFinished"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isSplitOn() {
+    throw new UnsupportedOperationException("isSplitOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isTableAvailable(TableName arg0, byte[][] arg1) {
+    throw new UnsupportedOperationException("isTableAvailable"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<ServerName>> listDeadServers() {
+    throw new UnsupportedOperationException("listDeadServers"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<ServerName>> listDecommissionedRegionServers() {
+    throw new UnsupportedOperationException("listDecommissionedRegionServers"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<NamespaceDescriptor>> listNamespaceDescriptors() {
+    throw new UnsupportedOperationException("abortProcedure"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<TableCFs>> listReplicatedTableCFs() {
+    throw new UnsupportedOperationException("listReplicatedTableCFs"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(
+      Optional<Pattern> arg0) {
+    throw new UnsupportedOperationException("listReplicationPeers"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<SnapshotDescription>> listSnapshots(Optional<Pattern> arg0) {
+    throw new UnsupportedOperationException("listSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern arg0,
+      Pattern arg1) {
+    throw new UnsupportedOperationException("listTableSnapshots"); // TODO ?
+  }
+
+  @Override
+  public CompletableFuture<Void> majorCompact(TableName arg0, Optional<byte[]> arg1) {
+    throw new UnsupportedOperationException("majorCompact"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> majorCompactRegion(byte[] arg0, Optional<byte[]> arg1) {
+    throw new UnsupportedOperationException("majorCompactRegion"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> majorCompactRegionServer(ServerName arg0) {
+    throw new UnsupportedOperationException("majorCompactRegionServer"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> mergeRegions(byte[] arg0, byte[] arg1, boolean arg2) {
+    throw new UnsupportedOperationException("mergeRegions"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> modifyColumnFamily(TableName arg0, ColumnFamilyDescriptor arg1) {
+    throw new UnsupportedOperationException("modifyColumnFamily"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> modifyNamespace(NamespaceDescriptor arg0) {
+    throw new UnsupportedOperationException("modifyNamespace"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> move(byte[] arg0, Optional<ServerName> arg1) {
+    throw new UnsupportedOperationException("move"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> normalize() {
+    throw new UnsupportedOperationException("normalize"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> offline(byte[] arg0) {
+    throw new UnsupportedOperationException("offline"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> recommissionRegionServer(ServerName arg0, List<byte[]> arg1) {
+    throw new UnsupportedOperationException("recommissionRegionServer"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> removeReplicationPeer(String arg0) {
+    throw new UnsupportedOperationException("removeReplicationPeer"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> removeReplicationPeerTableCFs(String arg0,
+      Map<TableName, ? extends Collection<String>> arg1) {
+    throw new UnsupportedOperationException("removeReplicationPeerTableCFs"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> restoreSnapshot(String arg0) {
+    throw new UnsupportedOperationException("restoreSnapshot"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> restoreSnapshot(String arg0, boolean arg1) {
+    throw new UnsupportedOperationException("restoreSnapshot"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> rollWALWriter(ServerName arg0) {
+    throw new UnsupportedOperationException("rollWALWriter"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Integer> runCatalogJanitor() {
+    throw new UnsupportedOperationException("runCatalogJanitor"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> runCleanerChore() {
+    throw new UnsupportedOperationException("runCleanerChore"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> setBalancerOn(boolean arg0) {
+    throw new UnsupportedOperationException("setBalancerOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> setCatalogJanitorOn(boolean arg0) {
+    throw new UnsupportedOperationException("setCatalogJanitorOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> setCleanerChoreOn(boolean arg0) {
+    throw new UnsupportedOperationException("abosetCleanerChoreOnrtProcedure"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> setMergeOn(boolean arg0) {
+    throw new UnsupportedOperationException("setMergeOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> setNormalizerOn(boolean arg0) {
+    throw new UnsupportedOperationException("setNormalizerOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> setQuota(QuotaSettings arg0) {
+    throw new UnsupportedOperationException("setQuota"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> setSplitOn(boolean arg0) {
+    throw new UnsupportedOperationException("setSplitOn"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> shutdown() {
+    throw new UnsupportedOperationException("shutdown"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> snapshot(SnapshotDescription arg0) {
+    throw new UnsupportedOperationException("snapshot"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> split(TableName arg0) {
+    throw new UnsupportedOperationException("split"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> split(TableName arg0, byte[] arg1) {
+    throw new UnsupportedOperationException("split"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> splitRegion(byte[] arg0, Optional<byte[]> arg1) {
+    throw new UnsupportedOperationException("splitRegion"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> stopMaster() {
+    throw new UnsupportedOperationException("stopMaster"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> stopRegionServer(ServerName arg0) {
+    throw new UnsupportedOperationException("stopRegionServer"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> truncateTable(TableName arg0, boolean arg1) {
+    throw new UnsupportedOperationException("truncateTable"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> unassign(byte[] arg0, boolean arg1) {
+    throw new UnsupportedOperationException("unassign"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> updateConfiguration() {
+    throw new UnsupportedOperationException("updateConfiguration"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> updateConfiguration(ServerName arg0) {
+    throw new UnsupportedOperationException("updateConfiguration"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> updateReplicationPeerConfig(String arg0,
+      ReplicationPeerConfig arg1) {
+    throw new UnsupportedOperationException("updateReplicationPeerConfig");
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncBufferedMutator.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncBufferedMutator.java
new file mode 100644
index 000000000..cb4f0036a
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncBufferedMutator.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2017 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.hbase2_x;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.AsyncBufferedMutator;
+import org.apache.hadoop.hbase.client.Mutation;
+
+/**
+ * Bigtable implementation of {@link AsyncBufferedMutator}
+ * 
+ * @author spollapally
+ */
+public class BigtableAsyncBufferedMutator implements AsyncBufferedMutator {
+
+  @Override
+  public void close() {
+    throw new UnsupportedOperationException("close"); // TODO
+  }
+
+  @Override
+  public void flush() {
+    throw new UnsupportedOperationException("flush"); // TODO
+  }
+
+  @Override
+  public Configuration getConfiguration() {
+    throw new UnsupportedOperationException("getConfiguration"); // TODO
+  }
+
+  @Override
+  public TableName getName() {
+    throw new UnsupportedOperationException("getName"); // TODO
+  }
+
+  @Override
+  public long getWriteBufferSize() {
+    throw new UnsupportedOperationException("getWriteBufferSize"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> mutate(Mutation arg0) {
+    throw new UnsupportedOperationException("mutate"); // TODO
+  }
+
+  @Override
+  public List<CompletableFuture<Void>> mutate(List<? extends Mutation> arg0) {
+    throw new UnsupportedOperationException("mutate"); // TODO
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
new file mode 100644
index 000000000..c12c210b0
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
@@ -0,0 +1,177 @@
+/*
+ * Copyright 2017 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.hbase2_x;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.AsyncTable;
+import org.apache.hadoop.hbase.client.BigtableAsyncConnection;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.ScanResultConsumer;
+import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.grpc.BigtableDataClient;
+import com.google.cloud.bigtable.grpc.BigtableSession;
+import com.google.cloud.bigtable.hbase.BatchExecutor;
+import com.google.cloud.bigtable.hbase.adapters.HBaseRequestAdapter;
+
+/**
+ * Bigtable implementation of {@link AsyncTable}.
+ * 
+ * @author spollapally
+ */
+@SuppressWarnings("deprecation")
+public class BigtableAsyncTable implements AsyncTable{
+  private final Logger LOG = new Logger(getClass());
+
+  private final BigtableAsyncConnection asyncConnection;
+  private final BigtableDataClient client;
+  private final HBaseRequestAdapter hbaseAdapter;
+  private final TableName tableName;
+  private BatchExecutor batchExecutor;
+  
+  public BigtableAsyncTable(BigtableAsyncConnection asyncConnection,
+      HBaseRequestAdapter hbaseAdapter) {
+    this.asyncConnection = asyncConnection;
+    BigtableSession session = asyncConnection.getSession();
+    this.client = session.getDataClient();
+    this.hbaseAdapter = hbaseAdapter;
+    this.tableName = hbaseAdapter.getTableName();
+  }
+  
+  protected synchronized BatchExecutor getBatchExecutor() {
+    if (batchExecutor == null) {
+      batchExecutor = new BatchExecutor(asyncConnection.getSession(), hbaseAdapter);
+    }
+    return batchExecutor;
+  }
+  
+  @Override
+  public CompletableFuture<Result> append(Append append) {
+    throw new UnsupportedOperationException("append"); // TODO
+  }
+
+  @Override
+  public <T> List<CompletableFuture<T>> batch(List<? extends Row> actions) {
+    throw new UnsupportedOperationException("batch"); // TODO
+  }
+
+  @Override
+  public CheckAndMutateBuilder checkAndMutate(byte[] rowParam, byte[] familyParam) {
+    throw new UnsupportedOperationException("checkAndMutate"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> delete(Delete delete) {
+    throw new UnsupportedOperationException("delete"); // TODO
+  }
+
+  @Override
+  public List<CompletableFuture<Void>> delete(List<Delete> deletes) {
+    throw new UnsupportedOperationException("delete list"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Result> get(Get get) {
+    throw new UnsupportedOperationException("get"); // TODO
+  }
+
+  @Override
+  public List<CompletableFuture<Result>> get(List<Get> arg0) {
+    throw new UnsupportedOperationException("get"); // TODO
+  }
+
+  @Override
+  public Configuration getConfiguration() {
+    return this.asyncConnection.getConfiguration(); // TODO
+  }
+
+  @Override
+  public TableName getName() {
+    return this.tableName;
+  }
+
+  @Override
+  public long getOperationTimeout(TimeUnit timeUnit) {
+    throw new UnsupportedOperationException("getOperationTimeout"); // TODO
+  }
+
+  @Override
+  public long getReadRpcTimeout(TimeUnit arg0) {
+    throw new UnsupportedOperationException("getReadRpcTimeout"); // TODO
+  }
+
+  @Override
+  public long getRpcTimeout(TimeUnit arg0) {
+    throw new UnsupportedOperationException("getRpcTimeout"); // TODO
+  }
+
+  @Override
+  public long getScanTimeout(TimeUnit arg0) {
+    throw new UnsupportedOperationException("getScanTimeout"); // TODO
+  }
+
+  @Override
+  public long getWriteRpcTimeout(TimeUnit arg0) {
+    throw new UnsupportedOperationException("getWriteRpcTimeout"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Result> increment(Increment arg0) {
+    throw new UnsupportedOperationException("increment"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> mutateRow(RowMutations arg0) {
+    throw new UnsupportedOperationException("mutateRow"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> put(Put arg0) {
+    throw new UnsupportedOperationException("put"); // TODO
+  }
+
+  @Override
+  public List<CompletableFuture<Void>> put(List<Put> arg0) {
+    throw new UnsupportedOperationException("put"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<Result>> scanAll(Scan arg0) {
+    throw new UnsupportedOperationException("scanAll"); // TODO
+  }
+
+  @Override
+  public ResultScanner getScanner(Scan arg0) {
+    throw new UnsupportedOperationException("getScanner"); // TODO
+  }
+
+  @Override
+  public void scan(Scan arg0, ScanResultConsumer arg1) {
+    throw new UnsupportedOperationException("scan"); // TODO
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTableBuilderBase.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTableBuilderBase.java
new file mode 100644
index 000000000..8406bdd46
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTableBuilderBase.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2017 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.hbase2_x;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.hbase.client.AsyncTable;
+import org.apache.hadoop.hbase.client.AsyncTableBuilder;
+import org.apache.hadoop.hbase.client.RawAsyncTable;
+
+/**
+ * Abstract class for implementing Bigtable version of {@link AsyncTable} and {@link RawAsyncTable}
+ * 
+ * @author spollapally
+ */
+public abstract class BigtableAsyncTableBuilderBase<T extends AsyncTable>
+    implements AsyncTableBuilder<T> {
+
+  public BigtableAsyncTableBuilderBase() {
+  }
+  
+  @Override
+  public AsyncTableBuilder<T> setMaxAttempts(int arg0) {
+    throw new UnsupportedOperationException("setMaxAttempts"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<T> setOperationTimeout(long arg0, TimeUnit arg1) {
+    throw new UnsupportedOperationException("setOperationTimeout"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<T> setReadRpcTimeout(long arg0, TimeUnit arg1) {
+    throw new UnsupportedOperationException("setReadRpcTimeout"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<T> setRetryPause(long arg0, TimeUnit arg1) {
+    throw new UnsupportedOperationException("setRetryPause"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<T> setRpcTimeout(long arg0, TimeUnit arg1) {
+    throw new UnsupportedOperationException("setRpcTimeout"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<T> setScanTimeout(long arg0, TimeUnit arg1) {
+    throw new UnsupportedOperationException("setScanTimeout"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<T> setStartLogErrorsCnt(int arg0) {
+    throw new UnsupportedOperationException("setStartLogErrorsCnt"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<T> setWriteRpcTimeout(long arg0, TimeUnit arg1) {
+    throw new UnsupportedOperationException("setWriteRpcTimeout"); // TODO
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTableRegionLocator.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTableRegionLocator.java
new file mode 100644
index 000000000..d86b76807
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTableRegionLocator.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2017 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.hbase2_x;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.AsyncTableRegionLocator;
+import com.google.cloud.bigtable.config.BigtableOptions;
+import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.grpc.BigtableDataClient;
+import com.google.cloud.bigtable.grpc.BigtableTableName;
+import com.google.cloud.bigtable.hbase.adapters.SampledRowKeysAdapter;
+
+/**
+ * Bigtable implementation of {@link AsyncTableRegionLocator}
+ * 
+ * @author spollapally
+ */
+public class BigtableAsyncTableRegionLocator implements AsyncTableRegionLocator {
+  private final Logger LOG = new Logger(getClass());
+
+  private final TableName tableName;
+  private final BigtableDataClient client;
+  private final SampledRowKeysAdapter adapter;
+  private final BigtableTableName bigtableTableName;
+
+  public BigtableAsyncTableRegionLocator(TableName tableName, BigtableOptions options,
+      BigtableDataClient client) {
+    this.tableName = tableName;
+    this.client = client;
+    this.bigtableTableName = options.getInstanceName().toTableName(tableName.getNameAsString());
+    ServerName serverName = ServerName.valueOf(options.getDataHost(), options.getPort(), 0);
+    this.adapter = new SampledRowKeysAdapter(tableName, serverName) {
+
+      @Override
+      protected HRegionLocation createHRegionLocation(HRegionInfo hRegionInfo,
+          ServerName serverName) {
+        return new HRegionLocation(hRegionInfo, serverName);
+      }
+    };
+  }
+
+  @Override
+  public TableName getName() {
+    return this.tableName;
+  }
+
+  @Override
+  public CompletableFuture<HRegionLocation> getRegionLocation(byte[] arg0, boolean arg1) {
+    throw new UnsupportedOperationException("getRegionLocation"); // TODO
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableConnection.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableConnection.java
index a86c40f5f..ea006ec76 100644
--- a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableConnection.java
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableConnection.java
@@ -53,9 +53,10 @@ public class BigtableConnection extends AbstractBigtableConnection {
   }
 
   /**
-   * Due to hbase 1.x to 2.x binary incompatibilities. {@link HRegionLocation(HRegionInfo
-   * hRegionInfo)} will fail with NoSuchMethodException if not recompiled with hbase 2.0
-   * dependencies. Hence the override. See {@link SampledRowKeysAdapter} for more details.
+   * Due to hbase 1.x to 2.x binary incompatibilities.
+   * {@link HRegionLocation#HRegionLocation(org.apache.hadoop.hbase.client.RegionInfo, ServerName)}
+   * will fail with NoSuchMethodException if not recompiled with hbase 2.0 dependencies. Hence the
+   * override. See {@link SampledRowKeysAdapter} for more details.
    */
   @Override
   protected SampledRowKeysAdapter createSampledRowKeysAdapter(TableName tableName,
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/adapters/admin/TableAdapter2x.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/adapters/admin/TableAdapter2x.java
new file mode 100644
index 000000000..b94fecebb
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/adapters/admin/TableAdapter2x.java
@@ -0,0 +1,67 @@
+/*
+ * 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.hbase2_x.adapters.admin;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import com.google.bigtable.admin.v2.ColumnFamily;
+import com.google.bigtable.admin.v2.Table;
+import com.google.cloud.bigtable.config.BigtableOptions;
+import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.hbase.adapters.admin.ColumnDescriptorAdapter;
+import com.google.cloud.bigtable.hbase.adapters.admin.TableAdapter;
+
+/**
+ * Need this extended class as {@link TableAdapter#adapt(org.apache.hadoop.hbase.HTableDescriptor)}
+ * is not binary compatible with {@link TableAdapter2x#adapt(TableDescriptor)}
+ * 
+ * Similarly, {@link ColumnDescriptorAdapter#adapt(HColumnDescriptor)} is not binary compatible with
+ * {@link ColumnDescriptorAdapter#adapt(ColumnFamilyDescriptor)}. As of now, this class just copies
+ * the columnName into a HColumnDescriptor type, which seems to be sufficient for current test
+ * cases.
+ * 
+ * TODO: Current options are either a hacky cloning or extension with lot of duplicate code. Explore
+ * a better solution.
+ * 
+ * @author spollapally
+ */
+public class TableAdapter2x extends TableAdapter {
+  private final Logger LOG = new Logger(getClass());
+
+  public TableAdapter2x(BigtableOptions options, ColumnDescriptorAdapter columnDescriptorAdapter) {
+    super(options, columnDescriptorAdapter);
+  }
+
+  public Table adapt(TableDescriptor desc) {
+    Map<String, ColumnFamily> columnFamilies = new HashMap<>();
+    for (ColumnFamilyDescriptor column : desc.getColumnFamilies()) {
+      try {
+        String columnName = column.getNameAsString();
+        // TODO - This is a very shallow copy. See class comments for further details  
+        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(columnName);
+        ColumnFamily columnFamily = columnDescriptorAdapter.adapt(hColumnDescriptor).build();
+        columnFamilies.put(columnName, columnFamily);
+      } catch (Exception e) {
+        LOG.error("Failed to copy ColumnFamilyDescriptor to HColumnDescriptor", e);
+        throw new RuntimeException("Failed to copy ColumnFamilyDescriptor to HColumnDescriptor", e);
+      }
+    }
+    return Table.newBuilder().putAllColumnFamilies(columnFamilies).build();
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/org/apache/hadoop/hbase/client/BigtableAsyncConnection.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/org/apache/hadoop/hbase/client/BigtableAsyncConnection.java
new file mode 100644
index 000000000..4c3e0f001
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/main/java/org/apache/hadoop/hbase/client/BigtableAsyncConnection.java
@@ -0,0 +1,190 @@
+/*
+ * Copyright 2017 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 org.apache.hadoop.hbase.client;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.security.User;
+import com.google.cloud.bigtable.config.BigtableOptions;
+import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.grpc.BigtableSession;
+import com.google.cloud.bigtable.hbase.BigtableOptionsFactory;
+import com.google.cloud.bigtable.hbase.adapters.Adapters;
+import com.google.cloud.bigtable.hbase.adapters.HBaseRequestAdapter;
+import com.google.cloud.bigtable.hbase.adapters.HBaseRequestAdapter.MutationAdapters;
+import com.google.cloud.bigtable.hbase2_x.BigtableAsyncAdmin;
+
+/**
+ * Bigtable implementation of {@link AsyncConnection}
+ * 
+ * @author spollapally
+ */
+public class BigtableAsyncConnection implements AsyncConnection, Closeable {
+  private final Logger LOG = new Logger(getClass());
+  
+  private final Configuration conf;
+  private BigtableSession session;
+  private BigtableOptions options;
+  private volatile boolean closed = false;
+
+  private Set<TableName> disabledTables = new HashSet<>();
+  private MutationAdapters mutationAdapters;
+
+  static {
+    // This forces a clean class loading of both HConstants and KeyValue along
+    // with a whole bunch of other classes.
+    Adapters.class.getName();
+  }
+  
+  public BigtableAsyncConnection(Configuration conf) throws IOException {
+    this(conf, null, null, null);
+  }
+
+  public BigtableAsyncConnection(Configuration conf, AsyncRegistry registry, String clusterId,
+      User user) throws IOException {
+    LOG.debug("Creating BigtableAsyncConnection");
+    this.conf = conf;
+
+    BigtableOptions opts;
+    try {
+      opts = BigtableOptionsFactory.fromConfiguration(conf);
+    } catch (IOException ioe) {
+      LOG.error("Error loading BigtableOptions from Configuration.", ioe);
+      throw ioe;
+    }
+    
+    this.closed = false;
+    this.session = new BigtableSession(opts);
+    this.options = this.session.getOptions();
+  }
+
+  public HBaseRequestAdapter createAdapter(TableName tableName) {
+    if (mutationAdapters == null) {
+      synchronized (this) {
+        if (mutationAdapters == null) {
+          mutationAdapters = new HBaseRequestAdapter.MutationAdapters(options, conf);
+        }
+      }
+    }
+    return new HBaseRequestAdapter(options, tableName, mutationAdapters);
+  }
+
+  public BigtableSession getSession() {
+    return this.session;
+  }
+
+  public BigtableOptions getOptions() {
+    return this.options;
+  }
+  
+  public Set<TableName> getDisabledTables() {
+    return disabledTables;
+  }
+  
+  @Override
+  public void close() throws IOException {
+    LOG.debug("closeing BigtableAsyncConnection");
+    if (!this.closed) {
+      this.session.close();
+      this.closed = true;
+    }
+  }
+
+  @Override
+  public Configuration getConfiguration() {
+    return this.conf;
+  }
+
+  @Override
+  public AsyncAdminBuilder getAdminBuilder() {
+    return new AsyncAdminBuilder() {
+      
+      @Override
+      public AsyncAdminBuilder setStartLogErrorsCnt(int arg0) {
+        throw new UnsupportedOperationException("setStartLogErrorsCnt");
+      }
+      
+      @Override
+      public AsyncAdminBuilder setRpcTimeout(long arg0, TimeUnit arg1) {
+        throw new UnsupportedOperationException("setRpcTimeout");
+      }
+      
+      @Override
+      public AsyncAdminBuilder setRetryPause(long arg0, TimeUnit arg1) {
+        throw new UnsupportedOperationException("setRetryPause");
+      }
+      
+      @Override
+      public AsyncAdminBuilder setOperationTimeout(long arg0, TimeUnit arg1) {
+        throw new UnsupportedOperationException("setOperationTimeout");
+      }
+      
+      @Override
+      public AsyncAdminBuilder setMaxAttempts(int arg0) {
+        throw new UnsupportedOperationException("setMaxAttempts");
+      }
+      
+      @Override
+      public AsyncAdmin build() {
+        try {
+          return new BigtableAsyncAdmin(BigtableAsyncConnection.this);
+        } catch (IOException e) {
+          LOG.error("failed to build BigtableAsyncAdmin", e);
+          throw new UncheckedIOException("failed to build BigtableAsyncAdmin", e);
+        }
+      }
+    };
+  }
+
+  @Override
+  public AsyncAdminBuilder getAdminBuilder(ExecutorService arg0) {
+    throw new UnsupportedOperationException("getAdminBuilder ExecutorService"); // TODO
+  }
+
+  @Override
+  public AsyncBufferedMutatorBuilder getBufferedMutatorBuilder(TableName arg0) {
+    throw new UnsupportedOperationException("getBufferedMutatorBuilder"); // TODO
+  }
+
+  @Override
+  public AsyncBufferedMutatorBuilder getBufferedMutatorBuilder(TableName arg0,
+      ExecutorService arg1) {
+    throw new UnsupportedOperationException("getBufferedMutatorBuilder"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<RawAsyncTable> getRawTableBuilder(TableName arg0) {
+    throw new UnsupportedOperationException("getRawTableBuilder"); // TODO
+  }
+
+  @Override
+  public AsyncTableRegionLocator getRegionLocator(TableName tableName) {
+    throw new UnsupportedOperationException("getRegionLocator"); // TODO
+  }
+
+  @Override
+  public AsyncTableBuilder<AsyncTable> getTableBuilder(TableName tableName,
+      ExecutorService executorService) {
+    throw new UnsupportedOperationException("getTableBuilder"); // TODO
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/AppendAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/AppendAdapter.java
index 1e651f218..a6b4f0eb2 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/AppendAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/AppendAdapter.java
@@ -47,7 +47,7 @@ public class AppendAdapter implements OperationAdapter<Append, ReadModifyWriteRo
       List<Cell> cells = CellDeduplicationHelper.deduplicateFamily(operation, entry.getKey());
 
       for (Cell cell : cells) {
-        ReadModifyWriteRule.Builder rule = result.addRulesBuilder();
+        ReadModifyWriteRule.Builder rule = ReadModifyWriteRule.newBuilder();
         rule.setFamilyName(familyName);
         rule.setColumnQualifier(
             ByteString.copyFrom(
@@ -59,6 +59,7 @@ public class AppendAdapter implements OperationAdapter<Append, ReadModifyWriteRo
                 cell.getValueArray(),
                 cell.getValueOffset(),
                 cell.getValueLength()));
+        result.addRules(rule.build());
       }
     }
     return result;
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/DeleteAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/DeleteAdapter.java
index 5f7783e80..a96f282c7 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/DeleteAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/DeleteAdapter.java
@@ -62,7 +62,7 @@ public class DeleteAdapter extends MutationAdapter<Delete> {
         String.format("Cell type %s is unsupported.", cell.getTypeByte()));
   }
 
-  static void throwOnUnsupportedDeleteFamilyVersion(@SuppressWarnings("unused") Cell cell) {
+  static void throwOnUnsupportedDeleteFamilyVersion(Cell cell) {
     throw new UnsupportedOperationException(
         "Cannot perform column family deletion at timestamp.");
   }
@@ -113,9 +113,10 @@ public class DeleteAdapter extends MutationAdapter<Delete> {
         cell.getTimestamp(),
         BigtableConstants.HBASE_TIMEUNIT);
 
-      deleteBuilder.getTimeRangeBuilder()
+      deleteBuilder.setTimeRange(TimestampRange.newBuilder()
           .setStartTimestampMicros(startTimestamp)
-          .setEndTimestampMicros(endTimestamp);
+          .setEndTimestampMicros(endTimestamp)
+          .build());
     } else {
       // Delete all cells before a timestamp
       if (cell.getTimestamp() != HConstants.LATEST_TIMESTAMP) {
@@ -152,7 +153,8 @@ public class DeleteAdapter extends MutationAdapter<Delete> {
 
             mutations.add(Mutation.newBuilder()
               .setDeleteFromFamily(DeleteFromFamily.newBuilder()
-                  .setFamilyNameBytes(familyByteString).build())
+                  .setFamilyNameBytes(familyByteString)
+                  .build())
               .build());
           } else if (isFamilyVersionDelete(cell)) {
             throwOnUnsupportedDeleteFamilyVersion(cell);
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/IncrementAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/IncrementAdapter.java
index fde76be15..02aba86db 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/IncrementAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/IncrementAdapter.java
@@ -59,7 +59,7 @@ public class IncrementAdapter
           CellDeduplicationHelper.deduplicateFamily(operation, familyEntry.getKey());
 
       for (Cell cell : mutationCells){
-        ReadModifyWriteRule.Builder rule = result.addRulesBuilder();
+        ReadModifyWriteRule.Builder rule = ReadModifyWriteRule.newBuilder();
         rule.setIncrementAmount(
             Bytes.toLong(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
         rule.setFamilyName(familyName);
@@ -68,6 +68,7 @@ public class IncrementAdapter
                 cell.getQualifierArray(),
                 cell.getQualifierOffset(),
                 cell.getQualifierLength()));
+        result.addRules(rule.build());
       }
     }
     return result;
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java
index 3ac1db825..15000818a 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/PutAdapter.java
@@ -118,7 +118,8 @@ public class PutAdapter extends MutationAdapter<Put> {
                 .setFamilyNameBytes(familyString)
                 .setColumnQualifier(cellQualifierByteString)
                 .setValue(value)
-                .setTimestampMicros(timestampMicros))
+                .setTimestampMicros(timestampMicros)
+                .build())
             .build());
       }
     }
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/admin/ColumnDescriptorAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/admin/ColumnDescriptorAdapter.java
index 51c2fa066..7a8af1aa3 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/admin/ColumnDescriptorAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/admin/ColumnDescriptorAdapter.java
@@ -18,9 +18,9 @@ package com.google.cloud.bigtable.hbase.adapters.admin;
 import com.google.bigtable.admin.v2.ColumnFamily;
 import com.google.bigtable.admin.v2.GcRule;
 import com.google.bigtable.admin.v2.GcRule.Intersection;
-import com.google.bigtable.admin.v2.GcRule.Intersection.Builder;
 import com.google.bigtable.admin.v2.GcRule.RuleCase;
 import com.google.bigtable.admin.v2.GcRule.Union;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
@@ -31,6 +31,7 @@ import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HConstants;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -178,13 +179,12 @@ public class ColumnDescriptorAdapter {
       if (maxVersions == Integer.MAX_VALUE) {
         return null;
       } else {
-        return createMaxVersionsRule(maxVersions);
+        return maxVersions(maxVersions);
       }
     }
 
     // minVersions only comes into play with a TTL:
-    GcRule ageRule =
-        GcRule.newBuilder().setMaxAge(Duration.newBuilder().setSeconds(ttlSeconds)).build();
+    GcRule ageRule = maxAge(ttlSeconds);
     if (minVersions != HColumnDescriptor.DEFAULT_MIN_VERSIONS) {
       // The logic here is: only delete a cell if:
       //  1) the age is older than :ttlSeconds AND
@@ -194,20 +194,40 @@ public class ColumnDescriptorAdapter {
       // Intersection (AND)
       //    - maxAge = :HBase_ttlSeconds
       //    - maxVersions = :HBase_minVersion
-      GcRule minVersionRule = createMaxVersionsRule(minVersions);
-      Builder ageAndMin = Intersection.newBuilder().addRules(ageRule).addRules(minVersionRule);
-      ageRule = GcRule.newBuilder().setIntersection(ageAndMin).build();
+      ageRule = intersection(ageRule, maxVersions(minVersions));
     }
     if (maxVersions == Integer.MAX_VALUE) {
       return ageRule;
     } else {
-      GcRule maxVersionsRule = createMaxVersionsRule(maxVersions);
-      Union ageAndMax = Union.newBuilder().addRules(ageRule).addRules(maxVersionsRule).build();
-      return GcRule.newBuilder().setUnion(ageAndMax).build();
+      return union(ageRule, maxVersions(maxVersions));
     }
   }
 
-  private static GcRule createMaxVersionsRule(int maxVersions) {
+  @VisibleForTesting
+  static GcRule intersection(GcRule... rules) {
+    return GcRule.newBuilder()
+        .setIntersection(Intersection.newBuilder().addAllRules(Arrays.asList(rules)).build())
+        .build();
+  }
+
+  @VisibleForTesting
+  static GcRule union(GcRule... rules) {
+    return GcRule.newBuilder()
+        .setUnion(Union.newBuilder().addAllRules(Arrays.asList(rules)).build())
+        .build();
+  }
+
+  private static Duration duration(int ttlSeconds) {
+    return Duration.newBuilder().setSeconds(ttlSeconds).build();
+  }
+
+  @VisibleForTesting
+  static GcRule maxAge(int ttlSeconds) {
+    return GcRule.newBuilder().setMaxAge(duration(ttlSeconds)).build();
+  }
+
+  @VisibleForTesting
+  static GcRule maxVersions(int maxVersions) {
     return GcRule.newBuilder().setMaxNumVersions(maxVersions).build();
   }
 
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java
index 494c31754..17592cd98 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/adapters/read/GetAdapter.java
@@ -17,7 +17,6 @@ package com.google.cloud.bigtable.hbase.adapters.read;
 
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.RowSet;
-import com.google.bigtable.v2.ReadRowsRequest.Builder;
 import com.google.protobuf.ByteString;
 
 import org.apache.hadoop.hbase.client.Get;
@@ -44,7 +43,7 @@ public class GetAdapter implements ReadOperationAdapter<Get> {
 
   /** {@inheritDoc} */
   @Override
-  public Builder adapt(Get operation, ReadHooks readHooks) {
+  public ReadRowsRequest.Builder adapt(Get operation, ReadHooks readHooks) {
     Scan operationAsScan = new Scan(operation);
     scanAdapter.throwIfUnsupportedScan(operationAsScan);
     return ReadRowsRequest.newBuilder()


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestRetryOptions.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestRetryOptions.java
index 90a7182ae..a27b97d54 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestRetryOptions.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/config/TestRetryOptions.java
@@ -42,7 +42,6 @@ public class TestRetryOptions {
   private void assertIsRetryableRead(boolean retryOnDeadlineExceeded) {
     RetryOptions options =
         new RetryOptions.Builder().setRetryOnDeadlineExceeded(retryOnDeadlineExceeded).build();
-    assertTrue(options.isRetryable(Status.INTERNAL.getCode()));
     assertTrue(options.isRetryable(Status.UNAVAILABLE.getCode()));
     assertTrue(options.isRetryable(Status.ABORTED.getCode()));
     assertFalse(options.isRetryable(Status.FAILED_PRECONDITION.getCode()));
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestAppProfile.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestAppProfile.java
new file mode 100644
index 000000000..9c451efca
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestAppProfile.java
@@ -0,0 +1,304 @@
+/*
+ * Copyright 2016 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.bigtable.v2.BigtableGrpc.BigtableImplBase;
+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.MutateRowsResponse.Entry;
+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.bigtable.v2.RowSet;
+import com.google.bigtable.v2.SampleRowKeysRequest;
+import com.google.bigtable.v2.SampleRowKeysResponse;
+import com.google.cloud.bigtable.config.BigtableOptions;
+import com.google.cloud.bigtable.config.CredentialOptions;
+import com.google.cloud.bigtable.grpc.async.BulkMutation;
+import com.google.cloud.bigtable.grpc.async.BulkRead;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Queues;
+import com.google.protobuf.ByteString;
+import com.google.rpc.Code;
+import com.google.rpc.Status;
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.stub.StreamObserver;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class TestAppProfile {
+  private FakeDataService fakeDataService;
+  private Server server;
+
+  private BigtableSession defaultSession;
+  private BigtableSession profileSession;
+
+  @Before
+  public void setUp() throws IOException {
+    fakeDataService = new FakeDataService();
+
+    final int port;
+    try(ServerSocket s = new ServerSocket(0)) {
+      port = s.getLocalPort();
+    }
+    server = ServerBuilder.forPort(port)
+        .addService(fakeDataService)
+        .build();
+    server.start();
+
+
+    BigtableOptions opts = new BigtableOptions.Builder()
+        .setDataHost("localhost")
+        .setInstanceAdminHost("locahost")
+        .setTableAdminHost("localhost")
+        .setPort(port)
+        .setProjectId("fake-project")
+        .setInstanceId("fake-instance")
+        .setUserAgent("fake-agent")
+        .setUsePlaintextNegotiation(true)
+        .setCredentialOptions(CredentialOptions.nullCredential())
+        .build();
+
+    defaultSession = new BigtableSession(opts);
+
+    profileSession = new BigtableSession(
+        opts.toBuilder()
+        .setAppProfileId("my-app-profile")
+        .build()
+    );
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (defaultSession != null) {
+      defaultSession.close();
+    }
+
+    if (profileSession != null) {
+      profileSession.close();
+    }
+
+    if (server != null) {
+      server.shutdownNow();
+      server.awaitTermination();
+    }
+  }
+
+  @Test
+  public void testReadRows() throws Exception {
+    defaultSession.getDataClient().readRows(ReadRowsRequest.getDefaultInstance()).next();
+    ReadRowsRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    profileSession.getDataClient().readRows(ReadRowsRequest.getDefaultInstance());
+    ReadRowsRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+  }
+
+  @Test
+  public void testSampleRowKeys() throws Exception {
+    defaultSession.getDataClient().sampleRowKeys(SampleRowKeysRequest.getDefaultInstance());
+    SampleRowKeysRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    profileSession.getDataClient().sampleRowKeys(SampleRowKeysRequest.getDefaultInstance());
+    SampleRowKeysRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+  }
+
+  @Test
+  public void testMutateRow() throws Exception {
+    defaultSession.getDataClient().mutateRow(MutateRowRequest.getDefaultInstance());
+    MutateRowRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    profileSession.getDataClient().mutateRow(MutateRowRequest.getDefaultInstance());
+    MutateRowRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+
+  }
+
+  @Test
+  public void testMutateRows() throws Exception {
+    defaultSession.getDataClient().mutateRows(MutateRowsRequest.getDefaultInstance());
+    MutateRowsRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    profileSession.getDataClient().mutateRows(MutateRowsRequest.getDefaultInstance());
+    MutateRowsRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+  }
+
+  @Test
+  public void testCheckAndMutateRow() throws Exception {
+    defaultSession.getDataClient().checkAndMutateRow(CheckAndMutateRowRequest.getDefaultInstance());
+    CheckAndMutateRowRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    profileSession.getDataClient().checkAndMutateRow(CheckAndMutateRowRequest.getDefaultInstance());
+    CheckAndMutateRowRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+  }
+
+  @Test
+  public void testReadModifyWrite() throws Exception {
+    defaultSession.getDataClient().readModifyWriteRow(ReadModifyWriteRowRequest.getDefaultInstance());
+    ReadModifyWriteRowRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    profileSession.getDataClient().readModifyWriteRow(ReadModifyWriteRowRequest.getDefaultInstance());
+    ReadModifyWriteRowRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+  }
+
+  @Test
+  public void testBulkMutation() throws Exception {
+    BigtableTableName fakeTableName = new BigtableTableName(
+        "projects/fake-project/instances/fake-instance/tables/fake-table");
+
+    MutateRowsRequest.Entry fakeMutationEntry = MutateRowsRequest.Entry
+        .newBuilder()
+        .setRowKey(ByteString.copyFromUtf8("fake-key"))
+        .build();
+
+    BulkMutation bulkMutation = defaultSession.createBulkMutation(fakeTableName);
+    bulkMutation.add(fakeMutationEntry);
+    bulkMutation.flush();
+
+    MutateRowsRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    BulkMutation bulkMutation2 = profileSession.createBulkMutation(fakeTableName);
+    bulkMutation2.add(fakeMutationEntry);
+    bulkMutation2.flush();
+
+    MutateRowsRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+  }
+
+  @Test
+  public void testBulkRead() throws Exception {
+    BigtableTableName fakeTableName = new BigtableTableName(
+        "projects/fake-project/instances/fake-instance/tables/fake-table");
+
+    ReadRowsRequest readRowsRequest = ReadRowsRequest.newBuilder()
+        .setTableName(fakeTableName.toString())
+        .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("fake-key")))
+        .build();
+
+
+    BulkRead bulkRead = defaultSession.createBulkRead(fakeTableName);
+    bulkRead.add(readRowsRequest);
+    bulkRead.flush();
+
+    ReadRowsRequest req = fakeDataService.popLastRequest();
+    Preconditions.checkState(req.getAppProfileId().isEmpty());
+
+    BulkRead bulkRead2 = profileSession.createBulkRead(fakeTableName);
+    bulkRead2.add(readRowsRequest);
+    bulkRead2.flush();
+
+    ReadRowsRequest req2 = fakeDataService.popLastRequest();
+    Assert.assertEquals(req2.getAppProfileId(), "my-app-profile");
+  }
+
+  static class FakeDataService extends BigtableImplBase {
+    final BlockingQueue<Object> requests = Queues.newLinkedBlockingDeque();
+
+    @SuppressWarnings("unchecked")
+    <T> T popLastRequest() throws InterruptedException {
+      return (T)requests.poll(1, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void readRows(ReadRowsRequest request,
+        StreamObserver<ReadRowsResponse> responseObserver) {
+      requests.add(request);
+      responseObserver.onCompleted();
+    }
+
+    @Override
+    public void sampleRowKeys(SampleRowKeysRequest request,
+        StreamObserver<SampleRowKeysResponse> responseObserver) {
+
+      requests.add(request);
+      responseObserver.onCompleted();
+    }
+
+    @Override
+    public void mutateRow(MutateRowRequest request,
+        StreamObserver<MutateRowResponse> responseObserver) {
+
+      requests.add(request);
+      responseObserver.onNext(MutateRowResponse.getDefaultInstance());
+      responseObserver.onCompleted();
+    }
+
+    @Override
+    public void mutateRows(MutateRowsRequest request,
+        StreamObserver<MutateRowsResponse> responseObserver) {
+
+      requests.add(request);
+      MutateRowsResponse.Builder response = MutateRowsResponse.newBuilder();
+      for (int i = 0; i < request.getEntriesCount(); i++) {
+        response.addEntries(
+            Entry.newBuilder()
+                .setIndex(i)
+                .setStatus(Status.newBuilder().setCode(Code.OK.getNumber()).build())
+                .build()
+        );
+      }
+      responseObserver.onNext(response.build());
+      responseObserver.onCompleted();
+    }
+
+    @Override
+    public void checkAndMutateRow(CheckAndMutateRowRequest request,
+        StreamObserver<CheckAndMutateRowResponse> responseObserver) {
+
+      requests.add(request);
+      CheckAndMutateRowResponse.Builder response = CheckAndMutateRowResponse.newBuilder()
+          .setPredicateMatched(false);
+
+      responseObserver.onNext(response.build());
+      responseObserver.onCompleted();
+    }
+
+    @Override
+    public void readModifyWriteRow(ReadModifyWriteRowRequest request,
+        StreamObserver<ReadModifyWriteRowResponse> responseObserver) {
+
+      requests.add(request);
+      responseObserver.onNext(ReadModifyWriteRowResponse.getDefaultInstance());
+      responseObserver.onCompleted();
+    }
+  }
+}
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestWaitForReplication.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestWaitForReplication.java
new file mode 100644
index 000000000..1f4a0d5a0
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestWaitForReplication.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright 2017 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 java.util.concurrent.TimeoutException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import com.google.api.client.testing.util.MockBackOff;
+import com.google.bigtable.admin.v2.BigtableTableAdminGrpc;
+import com.google.bigtable.admin.v2.CheckConsistencyRequest;
+import com.google.bigtable.admin.v2.CheckConsistencyResponse;
+import com.google.bigtable.admin.v2.GenerateConsistencyTokenRequest;
+import com.google.bigtable.admin.v2.GenerateConsistencyTokenResponse;
+import com.google.cloud.bigtable.config.BigtableOptions;
+
+import io.grpc.Status;
+import io.grpc.testing.GrpcServerRule;
+import io.grpc.stub.StreamObserver;
+import io.grpc.StatusException;
+
+@RunWith(JUnit4.class)
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class TestWaitForReplication {
+
+  // Class that implements a consistency service that returns dummy tokens and returns consistent
+  // after a configurable number of calls.
+  private static class ConsistencyServiceImpl extends BigtableTableAdminGrpc.BigtableTableAdminImplBase {
+
+    // How many calls need to be done before tokens are consistent.
+    private int callsToConsistency = 0;
+
+    // Number of calls done so far.
+    private int calls = 0;
+
+    public void setCallsToConsistency(int callsToConsistency) {
+      this.callsToConsistency = callsToConsistency;
+    }
+
+    @Override
+    public void generateConsistencyToken(GenerateConsistencyTokenRequest request,
+        StreamObserver<GenerateConsistencyTokenResponse> responseObserver) {
+
+      // Generates a token like "TokenFor-projects/P/instances/I/tables/T"
+      GenerateConsistencyTokenResponse response =
+          GenerateConsistencyTokenResponse.newBuilder()
+              .setConsistencyToken("TokenFor-" + request.getName())
+              .build();
+
+      responseObserver.onNext(response);
+      responseObserver.onCompleted();
+    }
+
+    @Override
+    public void checkConsistency(CheckConsistencyRequest request,
+        StreamObserver<CheckConsistencyResponse> responseObserver) {
+      calls++;
+
+      // Verify that we get the tokens we expect.
+      if (!request.getConsistencyToken().equals("TokenFor-" + request.getName())) {
+        responseObserver.onError(new StatusException(Status.INVALID_ARGUMENT));
+        return;
+
+      }
+
+      CheckConsistencyResponse response = CheckConsistencyResponse.newBuilder()
+              .setConsistent(calls >= callsToConsistency)
+              .build();
+
+      responseObserver.onNext(response);
+      responseObserver.onCompleted();
+    }
+  }
+
+  @Rule
+  public final GrpcServerRule grpcServerRule = new GrpcServerRule();
+
+  private static final BigtableTableName TABLE_NAME =
+      new BigtableTableName("projects/SomeProject/instances/SomeInstance/tables/SomeTable");
+
+  private ConsistencyServiceImpl service;
+  private BigtableTableAdminGrpcClient tableAdminClient;
+  private MockBackOff backoff;
+
+  @Before
+  public void setup() {
+    service = new ConsistencyServiceImpl();
+    grpcServerRule.getServiceRegistry().addService(service);
+
+    BigtableOptions options = new BigtableOptions.Builder().build();
+
+    tableAdminClient = new BigtableTableAdminGrpcClient(grpcServerRule.getChannel(), null, options);
+    backoff = new MockBackOff();
+  }
+
+  @Test
+  public void testConsistencyFast() throws Exception {
+    // Consistent right away.
+    service.setCallsToConsistency(1);
+    backoff.setMaxTries(10);
+    tableAdminClient.waitForReplication(TABLE_NAME, backoff);
+    Assert.assertEquals(0, backoff.getNumberOfTries());
+  }
+
+  @Test
+  public void testConsistencySlow() throws Exception {
+    // Consistent after a few calls.
+    service.setCallsToConsistency(10);
+    backoff.setMaxTries(9);
+    tableAdminClient.waitForReplication(TABLE_NAME, backoff);
+    Assert.assertEquals(9, backoff.getNumberOfTries());
+  }
+
+  @Test(expected = TimeoutException.class)
+  public void testConsistencyTimeOut() throws Exception {
+    // Time outs before consistency.
+    service.setCallsToConsistency(10);
+    backoff.setMaxTries(8);
+    tableAdminClient.waitForReplication(TABLE_NAME, backoff);
+  }
+}
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestMutateRowsRequestManager.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestMutateRowsRequestManager.java
new file mode 100644
index 000000000..5988ce175
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/async/TestMutateRowsRequestManager.java
@@ -0,0 +1,242 @@
+/*
+ * Copyright 2017 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.async;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.MockitoAnnotations;
+
+import com.google.api.client.util.NanoClock;
+import com.google.bigtable.v2.MutateRowsRequest;
+import com.google.bigtable.v2.MutateRowsRequest.Entry;
+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.config.RetryOptionsUtil;
+import com.google.cloud.bigtable.grpc.async.MutateRowsRequestManager.ProcessingStatus;
+import com.google.rpc.Status;
+
+import io.grpc.Status.Code;
+
+/**
+ * Tests for {@link MutateRowsRequestManager}
+ */
+@RunWith(JUnit4.class)
+public class TestMutateRowsRequestManager {
+
+  private static Status OK = statusOf(io.grpc.Status.Code.OK);
+  private static Status DEADLINE_EXCEEDED = statusOf(io.grpc.Status.Code.DEADLINE_EXCEEDED);
+  private static Status NOT_FOUND = statusOf(io.grpc.Status.Code.NOT_FOUND);
+
+  private static MutateRowsRequest createRequest(int entryCount) {
+    MutateRowsRequest.Builder builder = MutateRowsRequest.newBuilder();
+    for (int i = 0; i < entryCount; i++) {
+      Mutation mutation = Mutation.newBuilder()
+          .setSetCell(SetCell.newBuilder().setFamilyName("Family" + i).build()).build();
+      builder.addEntries(Entry.newBuilder().addMutations(mutation));
+    }
+    return builder.build();
+  }
+
+  private static MutateRowsRequest createRequest(MutateRowsRequest original, int ... indices) {
+    MutateRowsRequest.Builder builder = MutateRowsRequest.newBuilder();
+    for (int i : indices) {
+      builder.addEntries(original.getEntries(i));
+    }
+    return builder.build();
+  }
+
+  private static MutateRowsResponse createResponse(Status... statuses) {
+    MutateRowsResponse.Builder builder = MutateRowsResponse.newBuilder();
+    for (int i = 0; i < statuses.length; i++) {
+      builder.addEntries(toEntry(i, statuses[i]));
+    }
+    return builder.build();
+  }
+
+  private static com.google.bigtable.v2.MutateRowsResponse.Entry toEntry(int i, Status status) {
+    return MutateRowsResponse.Entry.newBuilder().setIndex(i).setStatus(status).build();
+  }
+
+  private static Status statusOf(Code code) {
+    return Status.newBuilder().setCode(code.value()).build();
+  }
+
+  private static ProcessingStatus send(MutateRowsRequestManager underTest,
+      MutateRowsResponse response) {
+    underTest.onMessage(response);
+    return underTest.onOK();
+  }
+
+  private AtomicLong time = new AtomicLong();
+  private NanoClock nanoClock = new NanoClock() {
+    @Override
+    public long nanoTime() {
+      return time.get();
+    }
+  };
+
+  private RetryOptions retryOptions;
+
+  @Before
+  public void setup() {
+    MockitoAnnotations.initMocks(this);
+    retryOptions = RetryOptionsUtil.createTestRetryOptions(nanoClock);
+  }
+
+  @Test
+  /**
+   * An empty request should return an empty response
+   */
+  public void testEmptySuccess() {
+    MutateRowsRequestManager underTest =
+        new MutateRowsRequestManager(retryOptions, createRequest(0));
+    send(underTest, createResponse());
+    Assert.assertEquals(createResponse(), underTest.buildResponse());
+  }
+
+  @Test
+  /**
+   * A single successful entry should work.
+   */
+  public void testSingleSuccess() {
+    MutateRowsRequestManager underTest =
+        new MutateRowsRequestManager(retryOptions, createRequest(1));
+    send(underTest, createResponse(OK));
+    Assert.assertEquals(createResponse(OK), underTest.buildResponse());
+  }
+
+  @Test
+  /**
+   * Two individual calls with one retry should work.
+   */
+  public void testTwoTrySuccessOneFailure() {
+    MutateRowsRequest originalRequest = createRequest(3);
+    MutateRowsRequestManager underTest =
+        new MutateRowsRequestManager(retryOptions, originalRequest);
+    send(underTest, createResponse(OK, DEADLINE_EXCEEDED, OK));
+    Assert.assertEquals(createRequest(originalRequest, 1), underTest.getRetryRequest());
+    Assert.assertEquals(createResponse(OK, DEADLINE_EXCEEDED, OK), underTest.buildResponse());
+    send(underTest, createResponse(OK));
+    Assert.assertEquals(createResponse(OK, OK, OK), underTest.buildResponse());
+  }
+
+  @Test
+  /**
+   * Two individual calls in a more complicated case with one retry should work.
+   */
+  public void testMultiSuccess() {
+    MutateRowsRequest originalRequest = createRequest(10);
+    MutateRowsRequestManager underTest =
+        new MutateRowsRequestManager(retryOptions, originalRequest);
+
+    // 5 mutations succeed, 5 mutations are retryable.
+    MutateRowsResponse firstResponse = createResponse(OK, DEADLINE_EXCEEDED, OK, DEADLINE_EXCEEDED,
+      OK, DEADLINE_EXCEEDED, OK, DEADLINE_EXCEEDED, OK, DEADLINE_EXCEEDED);
+    send(underTest, firstResponse);
+    Assert.assertEquals(createRequest(originalRequest, 1, 3, 5, 7, 9), underTest.getRetryRequest());
+    Assert.assertEquals(firstResponse, underTest.buildResponse());
+
+    // 3 mutations succeed, 2 mutations are retryable.
+    send(underTest, createResponse(OK, DEADLINE_EXCEEDED, OK, OK, DEADLINE_EXCEEDED));
+    Assert.assertEquals(createRequest(originalRequest, 3, 9), underTest.getRetryRequest());
+    MutateRowsResponse secondResponse = createResponse(OK, OK, OK, DEADLINE_EXCEEDED,
+      OK, OK, OK, OK, OK, DEADLINE_EXCEEDED);
+    Assert.assertEquals(secondResponse, underTest.buildResponse());
+
+    // The final 2 mutations are OK
+    send(underTest, createResponse(OK, OK));
+    Assert.assertEquals(createResponse(OK, OK, OK, OK, OK, OK, OK, OK, OK, OK),
+      underTest.buildResponse());
+  }
+
+  @Test
+  /**
+   * Multiple attempts at retries should work as expected. 10 mutations are added, and 1 gets an OK
+   * status for 9 rounds until 1 mutation is left.  Each success shows up in a random location.
+   */
+  public void testMultiAttempt() {
+    MutateRowsRequest originalRequest = createRequest(10);
+    MutateRowsRequestManager underTest =
+        new MutateRowsRequestManager(retryOptions, originalRequest);
+
+    // At the beginning, all mutations are outstanding
+    List<Integer> remaining = new ArrayList<>();
+    for (int i = 0; i < 10; i++) {
+      remaining.add(i);
+    }
+
+    for (int i = 0; i < 9; i++) {
+      int remainingMutationCount = remaining.size();
+      // Randomly choose a successful mutation
+      int successIndex = (int) (Math.random() * remainingMutationCount);
+
+      // Create a Status array filled with DEADLINE_EXCEEDED except for the chose one.
+      Status statuses[] = new Status[remainingMutationCount];
+      Arrays.fill(statuses, DEADLINE_EXCEEDED);
+      statuses[successIndex] = OK;
+
+      // The successful status can now be removed.
+      remaining.remove(successIndex);
+
+      // Make sure that the request is retryable, and that the retry request looks reasonable.
+      Assert.assertEquals(ProcessingStatus.RETRYABLE, send(underTest, createResponse(statuses)));
+      Assert.assertEquals(createRequest(originalRequest, toIntArray(remaining)),
+        underTest.getRetryRequest());
+    }
+
+    // Only one Mutation should be outstanding at this point. Create a response that has all OKs,
+    // with the exception of the remaining statuses.
+    Status[] statuses = new Status[10];
+    Arrays.fill(statuses, OK);
+    statuses[remaining.get(0)] = DEADLINE_EXCEEDED;
+    Assert.assertEquals(createResponse(statuses), underTest.buildResponse());
+  }
+
+  private final int[] toIntArray(List<Integer> integers) {
+    int indicesToRetry[] = new int[integers.size()];
+    for (int i = 0; i < integers.size(); i++) {
+      indicesToRetry[i] = integers.get(i);
+    }
+    return indicesToRetry;
+  }
+
+  @Test
+  public void testNotRetryable() {
+    MutateRowsRequestManager underTest =
+        new MutateRowsRequestManager(retryOptions, createRequest(3));
+    Assert.assertEquals(ProcessingStatus.NOT_RETRYABLE,
+      send(underTest, createResponse(OK, OK, NOT_FOUND)));
+    Assert.assertEquals(createResponse(OK, OK, NOT_FOUND), underTest.buildResponse());
+  }
+
+  @Test
+  public void testInvalid() {
+    // Create 3 muntations, but only 2 OKs.  That should be invalid
+    MutateRowsRequestManager underTest =
+        new MutateRowsRequestManager(retryOptions, createRequest(3));
+    Assert.assertEquals(ProcessingStatus.INVALID, send(underTest, createResponse(OK, OK)));
+  }
+}
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 d92079699..1286338eb 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
@@ -45,6 +45,7 @@ import com.google.api.client.util.Clock;
 import com.google.bigtable.v2.BigtableGrpc;
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.ReadRowsResponse;
+import com.google.bigtable.v2.ReadRowsResponse.Builder;
 import com.google.bigtable.v2.ReadRowsResponse.CellChunk;
 import com.google.bigtable.v2.RowRange;
 import com.google.bigtable.v2.RowSet;
@@ -73,18 +74,21 @@ public class RetryingReadRowsOperationTest {
       ReadRowsRequest.newBuilder()
           .setRows(RowSet.newBuilder().addRowRanges(RowRange.newBuilder()
               .setStartKeyClosed(ByteString.EMPTY).setEndKeyOpen(ByteString.EMPTY).build()))
+          .setRowsLimit(10)
           .build();
 
-  public static ReadRowsResponse buildResponse(ByteString key) throws UnsupportedEncodingException {
-    return ReadRowsResponse.newBuilder()
-        .addChunks(CellChunk.newBuilder()
-          .setRowKey(key)
-          .setFamilyName(StringValue.newBuilder().setValue("Family"))
-          .setQualifier(BytesValue.newBuilder().setValue(ByteString.copyFrom("qualifier", "UTF-8")))
-          .setValue(ByteString.copyFrom("value", "UTF-8"))
-          .setCommitRow(true)
-          .build())
-        .build();
+  public static ReadRowsResponse buildResponse(ByteString ... keys) throws UnsupportedEncodingException {
+    Builder builder = ReadRowsResponse.newBuilder();
+    for (ByteString key : keys) {
+      builder.addChunks(CellChunk.newBuilder()
+        .setRowKey(key)
+        .setFamilyName(StringValue.newBuilder().setValue("Family"))
+        .setQualifier(BytesValue.newBuilder().setValue(ByteString.copyFrom("qualifier", "UTF-8")))
+        .setValue(ByteString.copyFrom("value", "UTF-8"))
+        .setCommitRow(true)
+        .build());
+    }
+    return builder.build();
   }
 
   @Rule
@@ -165,13 +169,28 @@ public class RetryingReadRowsOperationTest {
     ReadRowsResponse response = buildResponse(key);
     underTest.onMessage(response);
     verify(mockFlatRowObserver, times(1)).onNext(any(FlatRow.class));
-    checkRetryRequest(underTest, key);
+    checkRetryRequest(underTest, key, 9);
     verify(mockClientCall, times(2)).request(eq(1));
 
     finishOK(underTest, 0);
   }
 
 
+  @Test
+  public void testDoubleResponse() throws UnsupportedEncodingException {
+    RetryingReadRowsOperation underTest = createOperation(mockFlatRowObserver);
+    start(underTest);
+    ByteString key1 = ByteString.copyFrom("SomeKey1", "UTF-8");
+    ByteString key2 = ByteString.copyFrom("SomeKey2", "UTF-8");
+    ReadRowsResponse response = buildResponse(key1, key2);
+    underTest.onMessage(response);
+    verify(mockFlatRowObserver, times(2)).onNext(any(FlatRow.class));
+    checkRetryRequest(underTest, key2, 8);
+    verify(mockClientCall, times(2)).request(eq(1));
+
+    finishOK(underTest, 0);
+  }
+
   @Test
   public void testSingleResponseWithQueue() throws UnsupportedEncodingException {
     ResponseQueueReader reader = new ResponseQueueReader();
@@ -180,7 +199,7 @@ public class RetryingReadRowsOperationTest {
     ByteString key = ByteString.copyFrom("SomeKey", "UTF-8");
     ReadRowsResponse response = buildResponse(key);
     underTest.onMessage(response);
-    checkRetryRequest(underTest, key);
+    checkRetryRequest(underTest, key, 9);
 
     verify(mockClientCall, times(1)).request(eq(1));
 
@@ -197,7 +216,7 @@ public class RetryingReadRowsOperationTest {
     underTest.onMessage(buildResponse(key1));
     underTest.onMessage(buildResponse(key2));
     verify(mockFlatRowObserver, times(2)).onNext(any(FlatRow.class));
-    checkRetryRequest(underTest, key2);
+    checkRetryRequest(underTest, key2, 8);
     verify(mockClientCall, times(3)).request(eq(1));
 
     finishOK(underTest, 0);
@@ -216,7 +235,7 @@ public class RetryingReadRowsOperationTest {
     Assert.assertNotSame(rw1, underTest.getRowMerger());
     underTest.onMessage(buildResponse(key2));
     verify(mockFlatRowObserver, times(2)).onNext(any(FlatRow.class));
-    checkRetryRequest(underTest, key2);
+    checkRetryRequest(underTest, key2, 8);
     verify(mockClientCall, times(4)).request(eq(1));
 
     finishOK(underTest, 1);
@@ -237,17 +256,17 @@ public class RetryingReadRowsOperationTest {
     performSuccessfulScanTimeouts(underTest, time);
     Assert.assertNotSame(rw1, underTest.getRowMerger());
     underTest.onClose(Status.ABORTED, new Metadata());
-    checkRetryRequest(underTest, key1);
+    checkRetryRequest(underTest, key1, 9);
 
     // a message gets through
     underTest.onMessage(buildResponse(key2));
     verify(mockFlatRowObserver, times(2)).onNext(any(FlatRow.class));
-    checkRetryRequest(underTest, key2);
+    checkRetryRequest(underTest, key2, 8);
 
     // more successful retries
     performSuccessfulScanTimeouts(underTest, time);
 
-    checkRetryRequest(underTest, key2);
+    checkRetryRequest(underTest, key2, 8);
     verify(mockClientCall, times(RETRY_OPTIONS.getMaxScanTimeoutRetries() * 2 + 4)).request(eq(1));
 
     // a succsesful finish.  There were 2 x RETRY_OPTIONS.getMaxScanTimeoutRetries() timeouts,
@@ -268,7 +287,7 @@ public class RetryingReadRowsOperationTest {
     // N successful scan timeout retries
     performSuccessfulScanTimeouts(underTest, time);
 
-    checkRetryRequest(underTest, key);
+    checkRetryRequest(underTest, key, 9);
 
     // one last scan timeout that fails.
     thrown.expect(BigtableRetriesExhaustedException.class);
@@ -290,14 +309,14 @@ public class RetryingReadRowsOperationTest {
     underTest.onClose(Status.ABORTED, new Metadata());
     Assert.assertNotNull(underTest.getCurrentBackoff());
     expectedRetryCount++;
-    checkRetryRequest(underTest, key1);
+    checkRetryRequest(underTest, key1, 9);
 
     // N successful scan timeout retries
     for (int i = 0; i < 2; i++) {
       performTimeout(underTest, time);
       expectedRetryCount++;
     }
-    checkRetryRequest(underTest, key1);
+    checkRetryRequest(underTest, key1, 9);
     Assert.assertNull(underTest.getCurrentBackoff());
     underTest.onMessage(buildResponse(key2));
 
@@ -362,8 +381,10 @@ public class RetryingReadRowsOperationTest {
     verify(mockRpcTimerContext, times(expectedRetryCount + 1)).close();
   }
 
-  private static void checkRetryRequest(RetryingReadRowsOperation underTest, ByteString key) {
-    Assert.assertEquals(key,
-      underTest.getRetryRequest().getRows().getRowRanges(0).getStartKeyOpen());
+  private static void checkRetryRequest(RetryingReadRowsOperation underTest, ByteString key,
+      int rowCount) {
+    ReadRowsRequest request = underTest.buildUpdatedRequst();
+    Assert.assertEquals(key, request.getRows().getRowRanges(0).getStartKeyOpen());
+    Assert.assertEquals(rowCount, request.getRowsLimit());
   }
 }
diff --git a/bigtable-hbase-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase2_x/FutureUtilsTest.java b/bigtable-hbase-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase2_x/FutureUtilsTest.java
index 9828b03f9..b5a4b3e69 100644
--- a/bigtable-hbase-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase2_x/FutureUtilsTest.java
+++ b/bigtable-hbase-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase2_x/FutureUtilsTest.java
@@ -99,8 +99,5 @@ public class FutureUtilsTest {
     thrown.expectCause(IsInstanceOf.<Throwable>instanceOf(IllegalStateException.class));
     thrown.expectMessage(containsString("Test failed feature"));
     completableFuture.get();
-    
-    assertTrue("future should be complete", completableFuture.isDone());
-    assertTrue("Should complete with exception", completableFuture.isCompletedExceptionally());
   }
 }
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/pom.xml b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/pom.xml
index e49693dbb..72a9fa215 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/pom.xml
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/pom.xml
@@ -33,6 +33,8 @@ limitations under the License.
     <properties>
         <hbase.version>${hbase.version.2.0}</hbase.version>
         <google.bigtable.connection.impl>com.google.cloud.bigtable.hbase2_x.BigtableConnection</google.bigtable.connection.impl>
+        <google.bigtable.async.connection.impl>org.apache.hadoop.hbase.client.BigtableAsyncConnection</google.bigtable.async.connection.impl>
+        <google.bigtable.registry.impl>org.apache.hadoop.hbase.client.BigtableAsyncRegistry</google.bigtable.registry.impl>
     </properties>
 
     <profiles>
@@ -65,6 +67,8 @@ limitations under the License.
                                     <reportNameSuffix>bigtable-server</reportNameSuffix>
                                     <systemPropertyVariables>
                                         <google.bigtable.connection.impl>${google.bigtable.connection.impl}</google.bigtable.connection.impl>
+                                        <google.bigtable.async.connection.impl>${google.bigtable.async.connection.impl}</google.bigtable.async.connection.impl>
+                                        <google.bigtable.registry.impl>${google.bigtable.registry.impl}</google.bigtable.registry.impl>
                                     </systemPropertyVariables>
                                     <forkedProcessTimeoutInSeconds>600</forkedProcessTimeoutInSeconds>
                                 </configuration>
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/AbstractTest.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/AbstractTest.java
index 815dd1668..58a1b87a4 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/AbstractTest.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/AbstractTest.java
@@ -17,6 +17,7 @@ package com.google.cloud.bigtable.hbase;
 
 import com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule;
 import java.io.IOException;
+import org.apache.hadoop.hbase.client.AsyncConnection;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.ClassRule;
@@ -49,6 +50,10 @@ public abstract class AbstractTest {
     return sharedTestEnv.getConnection();
   }
 
+  protected AsyncConnection getAsyncConnection() throws Exception {
+    return sharedTestEnv.getAsynConnection();
+  }
+
   protected static class QualifierValue implements Comparable<QualifierValue> {
 
     protected final byte[] qualifier;
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java
index af548f442..02ebac1a0 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/IntegrationTests.java
@@ -15,6 +15,7 @@
  */
 package com.google.cloud.bigtable.hbase;
 
+import com.google.cloud.bigtable.hbase.async.TestAsyncAdmin;
 import com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule;
 import java.util.concurrent.TimeUnit;
 import org.junit.ClassRule;
@@ -45,7 +46,8 @@ import org.junit.runners.Suite;
     TestPut.class,
     TestTimestamp.class,
     TestTruncateTable.class,
-    TestImport.class
+    TestImport.class,
+    TestAsyncAdmin.class
 })
 public class IntegrationTests {
 
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestCreateTable.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestCreateTable.java
index 8f053482a..2f62be35c 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestCreateTable.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestCreateTable.java
@@ -106,7 +106,7 @@ public class TestCreateTable extends AbstractTest {
       try {
         admin.createTable(new HTableDescriptor(TableName.valueOf(badName))
             .addFamily(new HColumnDescriptor(COLUMN_FAMILY)));
-      } catch (RuntimeException ex) {
+      } catch (Exception ex) {
         //TODO verify - added RuntimeException check as RandomStringUtils seems to be generating a string server side doesn't like 
         failed = true;
       }
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncAdmin.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncAdmin.java
new file mode 100644
index 000000000..e873ff357
--- /dev/null
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncAdmin.java
@@ -0,0 +1,207 @@
+/*
+ * Copyright 2017 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.hbase.async;
+
+import static com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule.COLUMN_FAMILY;
+import static org.junit.Assert.*;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNotEnabledException;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.AsyncAdmin;
+import org.apache.hadoop.hbase.client.AsyncConnection;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.hamcrest.core.IsInstanceOf;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import com.google.cloud.bigtable.hbase.AbstractTest;
+import com.google.cloud.bigtable.hbase.Logger;
+
+/**
+ * Integration tests for BigtableAsyncConnection
+ * 
+ * @author spollapally
+ */
+public class TestAsyncAdmin extends AbstractTest {
+  private final Logger LOG = new Logger(getClass());
+
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  @Test
+  public void testAsyncConnection() throws Exception {
+    AsyncConnection asyncCon = getAsyncConnection();
+    Assert.assertNotNull("async connection should not be null", asyncCon);
+    AsyncAdmin asycAdmin = asyncCon.getAdmin();
+    Assert.assertNotNull("asycAdmin should not be null", asycAdmin);
+  }
+
+  @Test
+  public void testCreateTable_exception() throws Exception {
+    AsyncAdmin asyncAdmin = getAsyncConnection().getAdmin();
+    thrown.expect(NullPointerException.class);
+    asyncAdmin.createTable(null).get();
+  }
+
+  @Test
+  public void testCreateTable_harness() throws Exception {
+    AsyncAdmin asyncAdmin = getAsyncConnection().getAdmin();
+    TableName tableName = TableName.valueOf("TestTable" + UUID.randomUUID().toString());
+
+    try {
+      // test exists
+      assertEquals(false, asyncAdmin.tableExists(tableName).get());
+
+      // test create new
+      asyncAdmin.createTable(TableDescriptorBuilder.newBuilder(tableName)
+          .addColumnFamily(ColumnFamilyDescriptorBuilder.of(COLUMN_FAMILY)).build()).get();
+      assertEquals(true, asyncAdmin.tableExists(tableName).get());
+
+      // test listTableNames all
+      List<TableName> allTableNames = asyncAdmin.listTableNames().get();
+      assertTrue("listTableNames-all should list atleast one table", allTableNames.size() > 0);
+      assertTrue("listTableNames-all Should contain tableName" + tableName,
+          allTableNames.stream().anyMatch(e -> tableName.equals(e)));
+
+      // test listTableNames by pattern
+      String tNStr = tableName.getNameAsString();
+      List<TableName> patTableNames = asyncAdmin
+          .listTableNames(Optional.of(Pattern.compile(tNStr.substring(0, 15).concat(".*"))), false)
+          .get();
+      assertTrue("listTableNames-pattern should list atleast one table", patTableNames.size() > 0);
+      assertTrue("listTableNames-pattern should contain tableName" + tableName,
+          patTableNames.stream().anyMatch(e -> tableName.equals(e)));
+
+      // test listTables all
+      List<TableDescriptor> allTableDescriptors = asyncAdmin.listTables().get();
+      assertTrue("listTables-all should list atleast one table", allTableDescriptors.size() > 0);
+      assertTrue("listTables-all should contain tableName" + tableName,
+          allTableDescriptors.stream().anyMatch(e -> tableName.equals(e.getTableName())));
+
+      // test listTables by pattern
+      List<TableDescriptor> patTableDescriptors = asyncAdmin
+          .listTables(Optional.of(Pattern.compile(tNStr.substring(0, 15).concat(".*"))), false)
+          .get();
+      assertTrue("listTables-pattern should list atleast one table",
+          allTableDescriptors.size() > 0);
+      assertTrue("listTables-pattern should contain tableName" + tableName,
+          patTableDescriptors.stream().anyMatch(e -> tableName.equals(e.getTableName())));
+
+      // test isTableEnabled
+      assertEquals(true, asyncAdmin.isTableEnabled(tableName).get());
+
+      // test disableTable and isTableDisabled
+      asyncAdmin.disableTable(tableName).get();
+      assertEquals(true, asyncAdmin.isTableDisabled(tableName).get());
+
+      // test deleteTable
+      asyncAdmin.deleteTable(tableName).get();
+      assertEquals(false, asyncAdmin.tableExists(tableName).get());
+    } finally {
+      // not strictly required. New test suite run for BigtableEnv will do cleanup by configured
+      // table name patterns
+      deleteTestTable(tableName);
+    }
+  }
+
+  @Test
+  public void testCreateTableWithNumRegions_exception() throws Exception {
+    AsyncAdmin asyncAdmin = getAsyncConnection().getAdmin();
+    TableName tableName = TableName.valueOf("TestTable" + UUID.randomUUID().toString());
+    thrown.expect(ExecutionException.class);
+    thrown.expectCause(IsInstanceOf.<Throwable>instanceOf(IllegalArgumentException.class));
+    asyncAdmin.createTable(TableDescriptorBuilder.newBuilder(tableName).build(),
+        Bytes.toBytes("AAA"), Bytes.toBytes("BBB"), 2).get();
+  }
+
+  @Test
+  public void testCreateTableWithSplits() throws Exception {
+    AsyncAdmin asyncAdmin = getAsyncConnection().getAdmin();
+    TableName tableName1 = TableName.valueOf("TestTable" + UUID.randomUUID().toString());
+    TableName tableName2 = TableName.valueOf("TestTable" + UUID.randomUUID().toString());
+
+    try {
+      asyncAdmin.createTable(
+          TableDescriptorBuilder.newBuilder(tableName1)
+              .addColumnFamily(ColumnFamilyDescriptorBuilder.of(COLUMN_FAMILY)).build(),
+          Bytes.toBytes("AAA"), Bytes.toBytes("BBB"), 3).get();
+      assertEquals(true, asyncAdmin.tableExists(tableName1).get());
+
+      byte[][] splitKeys =
+          new byte[][] {Bytes.toBytes("AAA"), Bytes.toBytes("BBB"), Bytes.toBytes("CCC"),};
+
+      asyncAdmin.createTable(
+          TableDescriptorBuilder.newBuilder(tableName2)
+              .addColumnFamily(ColumnFamilyDescriptorBuilder.of(COLUMN_FAMILY)).build(),
+          Optional.of(splitKeys)).get();
+      assertEquals(true, asyncAdmin.tableExists(tableName2).get());
+
+      // TODO - Add Region checks
+    } finally {
+      deleteTestTable(tableName1);
+      deleteTestTable(tableName2);
+    }
+  }
+
+  @Test
+  public void testDisbleTable_exceptions() throws Exception {
+    AsyncAdmin asyncAdmin = getAsyncConnection().getAdmin();
+    TableName tableName = TableName.valueOf("TestTable" + UUID.randomUUID().toString());
+
+    // test non existing table
+    Throwable th = null;
+    try {
+      asyncAdmin.disableTable(tableName).get();
+    } catch (ExecutionException e) {
+      th = e;
+    }
+    assertThat(th.getCause(), IsInstanceOf.<Throwable>instanceOf(TableNotFoundException.class));
+
+    // test already disabled table
+    th = null;
+    asyncAdmin.createTable(TableDescriptorBuilder.newBuilder(tableName)
+        .addColumnFamily(ColumnFamilyDescriptorBuilder.of(COLUMN_FAMILY)).build()).get();
+    asyncAdmin.disableTable(tableName).get();
+    try {
+      asyncAdmin.disableTable(tableName).get();
+    } catch (ExecutionException e) {
+      th = e;
+    }
+    assertThat(th.getCause(), IsInstanceOf.<Throwable>instanceOf(TableNotEnabledException.class));
+  }
+
+  private void deleteTestTable(TableName tableName) {
+    try {
+      Admin admin = getConnection().getAdmin();
+      if (admin.tableExists(tableName)) {
+        getConnection().getAdmin().disableTable(tableName);
+        getConnection().getAdmin().deleteTable(tableName);
+      }
+    } catch (Throwable t) {
+      LOG.warn("Error cleaning up the table", t);
+    }
+  }
+}
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java
index 02679d7d7..236c2e0e1 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java
@@ -26,18 +26,23 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.AsyncConnection;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 
 import com.google.bigtable.repackaged.io.grpc.internal.GrpcUtil;
 import com.google.cloud.bigtable.hbase.BigtableOptionsFactory;
+import com.google.cloud.bigtable.hbase.Logger;
 import com.google.common.collect.Sets;
 
 
 class BigtableEnv extends SharedTestEnv {
-
+  private final Logger LOG = new Logger(getClass());
+  
   private static final Set<String> KEYS = Sets.newHashSet(
       "hbase.client.connection.impl",
+      "hbase.client.async.connection.impl",
+      "hbase.client.registry.impl",
       BigtableOptionsFactory.BIGTABLE_PORT_KEY,
       BigtableOptionsFactory.BIGTABLE_HOST_KEY,
       BigtableOptionsFactory.BIGTABLE_INSTANCE_ADMIN_HOST_KEY,
@@ -53,12 +58,21 @@ class BigtableEnv extends SharedTestEnv {
   protected void setup() throws IOException {
     configuration = HBaseConfiguration.create();
 
-
     String connectionClass = System.getProperty("google.bigtable.connection.impl");
     if (connectionClass != null) {
       configuration.set("hbase.client.connection.impl", connectionClass);
     }
 
+    String asyncConnectionClass = System.getProperty("google.bigtable.async.connection.impl");
+    if (asyncConnectionClass != null) {
+      configuration.set("hbase.client.async.connection.impl", asyncConnectionClass);
+    }
+
+    String registryClass = System.getProperty("google.bigtable.registry.impl");
+    if (registryClass != null) {
+      configuration.set("hbase.client.registry.impl", registryClass);
+    }
+    
     for (Entry<Object, Object> entry : System.getProperties().entrySet()) {
       if (KEYS.contains(entry.getKey())) {
         configuration.set(entry.getKey().toString(), entry.getValue().toString());
@@ -68,12 +82,13 @@ class BigtableEnv extends SharedTestEnv {
     ExecutorService executor = Executors.newCachedThreadPool(GrpcUtil.getThreadFactory("table_deleter", true));
     try (Connection connection = createConnection();
         Admin admin = connection.getAdmin()) {
-      for (final TableName tableName : admin.listTableNames("(test_table|list_table[12])-.*")) {
+      for (final TableName tableName : admin.listTableNames("(test_table|list_table[12]|TestTable).*")) {
         executor.execute(new Runnable() {
           @Override
           public void run() {
             try {
               admin.deleteTable(tableName);
+              LOG.info("Test-setup deleting table: %s", tableName.getNameAsString());
             } catch (IOException e) {
               e.printStackTrace();
             }
@@ -98,4 +113,9 @@ class BigtableEnv extends SharedTestEnv {
   public Connection createConnection() throws IOException {
     return ConnectionFactory.createConnection(configuration);
   }
+  
+  public AsyncConnection createAsyncConnection() throws Exception {
+    return ConnectionFactory.createAsyncConnection(configuration).get();
+  }
+
 }
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/MiniClusterEnv.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/MiniClusterEnv.java
index ddaf26d3e..e4e0c71af 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/MiniClusterEnv.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/MiniClusterEnv.java
@@ -21,6 +21,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.client.AsyncConnection;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 
@@ -68,4 +69,19 @@ class MiniClusterEnv extends SharedTestEnv {
 
     return ConnectionFactory.createConnection(clientConfig);
   }
+
+  @Override
+  public AsyncConnection createAsyncConnection() throws Exception {
+    Configuration clientConfig = HBaseConfiguration.create();
+    
+    String[] keys = new String[]{
+        "hbase.zookeeper.quorum",
+        "hbase.zookeeper.property.clientPort"
+    };
+    for (String key : keys) {
+      clientConfig.set(key, helper.getConfiguration().get(key));
+    }
+
+    return ConnectionFactory.createAsyncConnection(clientConfig).get();
+  }
 }
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java
index a00d01f6a..c495b3d26 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java
@@ -17,6 +17,7 @@ package com.google.cloud.bigtable.hbase.test_env;
 
 import com.google.common.base.Preconditions;
 import java.io.IOException;
+import org.apache.hadoop.hbase.client.AsyncConnection;
 import org.apache.hadoop.hbase.client.Connection;
 
 /**
@@ -70,4 +71,6 @@ abstract class SharedTestEnv {
   protected abstract void teardown() throws IOException;
 
   public abstract Connection createConnection() throws IOException;
+  
+  public abstract AsyncConnection createAsyncConnection() throws Exception;
 }
diff --git a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java
index 35d6ca8da..4129a444a 100644
--- a/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java
+++ b/bigtable-hbase-parent/bigtable-hbase-integration-tests-2.x/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.AsyncConnection;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.rules.ExternalResource;
@@ -38,6 +39,7 @@ public class SharedTestEnvRule extends ExternalResource {
   private TableName defaultTableName;
   private SharedTestEnv sharedTestEnv;
   private Connection connection;
+  private AsyncConnection asyncConnection;
   private java.util.logging.Logger julLogger;
   private java.util.logging.Handler[] savedJulHandlers;
 
@@ -52,6 +54,7 @@ public class SharedTestEnvRule extends ExternalResource {
 
     sharedTestEnv = SharedTestEnv.get();
     connection = createConnection();
+    asyncConnection = createAsyncConnection();
 
     defaultTableName = newTestTableName();
     createTable(defaultTableName);
@@ -73,6 +76,14 @@ public class SharedTestEnvRule extends ExternalResource {
       LOG.error("Failed to close connection after test", e);
     }
     connection = null;
+
+    try {
+      asyncConnection.close();;
+    } catch (Exception e) {
+      LOG.error("Failed to close asyncConnection after test", e);
+    }
+    asyncConnection = null;
+    
     try {
       sharedTestEnv.release();
     } catch (IOException e) {
@@ -96,6 +107,14 @@ public class SharedTestEnvRule extends ExternalResource {
     return sharedTestEnv.createConnection();
   }
 
+  public AsyncConnection createAsyncConnection() throws Exception {
+    return sharedTestEnv.createAsyncConnection();
+  }
+
+  public AsyncConnection getAsynConnection() {
+    return asyncConnection;
+  }
+
   public boolean isBigtable() {
     // TODO(igorbernstein2): clean this up
     return sharedTestEnv instanceof BigtableEnv;
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/admin/TestColumnDescriptorAdapter.java b/bigtable-hbase-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/admin/TestColumnDescriptorAdapter.java
index 98d3c9963..ee7338871 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/admin/TestColumnDescriptorAdapter.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/adapters/admin/TestColumnDescriptorAdapter.java
@@ -15,7 +15,11 @@
  */
 package com.google.cloud.bigtable.hbase.adapters.admin;
 
-import java.util.Arrays;
+import static com.google.cloud.bigtable.hbase.adapters.admin.ColumnDescriptorAdapter.intersection;
+import static com.google.cloud.bigtable.hbase.adapters.admin.ColumnDescriptorAdapter.maxAge;
+import static com.google.cloud.bigtable.hbase.adapters.admin.ColumnDescriptorAdapter.maxVersions;
+import static com.google.cloud.bigtable.hbase.adapters.admin.ColumnDescriptorAdapter.union;
+
 import java.util.Map;
 import java.util.Set;
 
@@ -33,9 +37,6 @@ import org.junit.runners.JUnit4;
 
 import com.google.bigtable.admin.v2.ColumnFamily;
 import com.google.bigtable.admin.v2.GcRule;
-import com.google.bigtable.admin.v2.GcRule.Intersection;
-import com.google.bigtable.admin.v2.GcRule.Union;
-import com.google.protobuf.Duration;
 
 /**
  * Tests for {@link ColumnDescriptorAdapter}.
@@ -100,8 +101,7 @@ public class TestColumnDescriptorAdapter {
     descriptor.setScope(1); // REPLICATION_SCOPE
     descriptor.setInMemory(true);
 
-    ColumnFamily.Builder result = adapter.adapt(descriptor)
-        .clearGcRule();
+    ColumnFamily.Builder result = adapter.adapt(descriptor).clearGcRule();
 
     Assert.assertArrayEquals(
         new byte[0],
@@ -114,14 +114,14 @@ public class TestColumnDescriptorAdapter {
     int ttl = 86400;
     descriptor.setTimeToLive(ttl);
     ColumnFamily.Builder result = adapter.adapt(descriptor);
-    Assert.assertEquals(maxVersionsAndTtl(1, ttl), result.getGcRule());
+    Assert.assertEquals(union(maxAge(ttl), maxVersions(1)), result.getGcRule());
   }
 
   @Test
   public void ttlIsPreservedInColumnFamily() {
     // TTL of 1 day (in microseconds):
     HColumnDescriptor descriptor =
-        adapter.adapt("family", columnFamily(union(maxAgeRule(86400), maxNumVersionsRule(1))));
+        adapter.adapt("family", columnFamily(union(maxAge(86400), maxVersions(1))));
     Assert.assertEquals(1, descriptor.getMaxVersions());
     Assert.assertEquals(86400, descriptor.getTimeToLive());
   }
@@ -130,12 +130,12 @@ public class TestColumnDescriptorAdapter {
   public void maxVersionsIsPreservedInGcExpression() {
     descriptor.setMaxVersions(10);
     ColumnFamily.Builder result = adapter.adapt(descriptor);
-    Assert.assertEquals(maxNumVersionsRule(10), result.getGcRule());
+    Assert.assertEquals(maxVersions(10), result.getGcRule());
   }
 
   @Test
   public void maxVersionsIsPreservedInColumnFamily() {
-    HColumnDescriptor descriptor = adapter.adapt("family", columnFamily(maxNumVersionsRule(10)));
+    HColumnDescriptor descriptor = adapter.adapt("family", columnFamily(maxVersions(10)));
     Assert.assertEquals(10, descriptor.getMaxVersions());
   }
 
@@ -181,33 +181,12 @@ public class TestColumnDescriptorAdapter {
     return ColumnFamily.newBuilder().setGcRule(rule).build();
   }
 
-  private static GcRule maxNumVersionsRule(int maxNumVersions) {
-    return GcRule.newBuilder().setMaxNumVersions(maxNumVersions).build();
-  }
-
-  private static GcRule maxAgeRule(int ttlSecons) {
-    return GcRule.newBuilder().setMaxAge(Duration.newBuilder().setSeconds(ttlSecons)).build();
-  }
-
-  private static GcRule maxVersionsAndTtl(int maxVersions, int ttl) {
-    return union(maxAgeRule(ttl), maxNumVersionsRule(maxVersions));
-  }
-
   private GcRule minMaxRule(int minVersions, int ttl, int maxVersions) {
     return union(
         intersection(
-          maxAgeRule(ttl),
-          maxNumVersionsRule(minVersions)
+          maxAge(ttl),
+          maxVersions(minVersions)
          ),
-        maxNumVersionsRule(maxVersions));
-  }
-
-  private static GcRule union(GcRule... rules) {
-    return GcRule.newBuilder().setUnion(Union.newBuilder().addAllRules(Arrays.asList(rules))).build();
-  }
-
-  private static GcRule intersection(GcRule... rules) {
-    return GcRule.newBuilder()
-        .setIntersection(Intersection.newBuilder().addAllRules(Arrays.asList(rules))).build();
+        maxVersions(maxVersions));
   }
 }