Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1569 (2017-12-19 00:14:36)

app profiles


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/pom.xml b/bigtable-client-core-parent/bigtable-client-core/pom.xml
index 5792392cc..67a24ebf7 100644
--- a/bigtable-client-core-parent/bigtable-client-core/pom.xml
+++ b/bigtable-client-core-parent/bigtable-client-core/pom.xml
@@ -153,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>
@@ -193,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>
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/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/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 8e92ae188..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>
@@ -63,7 +63,8 @@ class ReadRowsRequestManager {
   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 4169c929e..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")
@@ -306,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
index ddb4226aa..324b75694 100644
--- 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
@@ -16,6 +16,7 @@
 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;
@@ -25,12 +26,16 @@ 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;
@@ -46,10 +51,20 @@ 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}
@@ -63,6 +78,7 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
   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");
@@ -70,44 +86,143 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     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) {
-    throw new UnsupportedOperationException("createTable"); // TODO
+    // 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) {
-    throw new UnsupportedOperationException("createTable"); // TODO
+
+    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) {
-    throw new UnsupportedOperationException("deleteTable"); // TODO
+    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) {
-    throw new UnsupportedOperationException("deleteTable"); // TODO
+    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) {
-    throw new UnsupportedOperationException("tableExists"); // TODO
+    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) {
-    throw new UnsupportedOperationException("listTables"); // TODO
+    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) {
-    throw new UnsupportedOperationException("listTables"); // TODO
+    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
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/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/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/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 73c9184ae..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
@@ -381,8 +381,9 @@ public class RetryingReadRowsOperationTest {
     verify(mockRpcTimerContext, times(expectedRetryCount + 1)).close();
   }
 
-  private static void checkRetryRequest(RetryingReadRowsOperation underTest, ByteString key, int rowCount) {
-    ReadRowsRequest request = underTest.getRetryRequest();
+  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/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
index 477df7c9f..e873ff357 100644
--- 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
@@ -15,27 +15,193 @@
  */
 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 
+ * 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 8321a8085..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
@@ -32,11 +32,13 @@ 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",
@@ -80,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();
             }
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));
   }
 }