Repository: googleapis/java-bigtable-hbase
Pull request #2592 (2020-07-16 15:11:47)

feat: managed backups implementation




Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/pom.xml b/bigtable-client-core-parent/bigtable-client-core/pom.xml
index 80a639452..bc5764250 100644
--- a/bigtable-client-core-parent/bigtable-client-core/pom.xml
+++ b/bigtable-client-core-parent/bigtable-client-core/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-client-core</artifactId>
@@ -151,6 +151,10 @@ limitations under the License.
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java-util</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>io.grpc</groupId>
@@ -283,6 +287,14 @@ this is not used directly, but upgrading due to transitive vulnerabilities in ol
           </usedDependencies>
         </configuration>
       </plugin>
+      <plugin>
+        <!-- skip for bigtable-client-core to allow for upcoming refactoring-->
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>clirr-maven-plugin</artifactId>
+        <configuration>
+          <skip>true</skip>
+        </configuration>
+      </plugin>
     </plugins>
   </build>
 </project>
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableTableAdminClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableTableAdminClient.java
index 45b416fa5..80a0500bd 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableTableAdminClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableTableAdminClient.java
@@ -24,9 +24,14 @@ import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsResponse;
 import com.google.bigtable.admin.v2.Snapshot;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.Backup;
+import com.google.cloud.bigtable.admin.v2.models.CreateBackupRequest;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoreTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoredTableResult;
 import com.google.cloud.bigtable.admin.v2.models.Table;
+import com.google.cloud.bigtable.admin.v2.models.UpdateBackupRequest;
 import com.google.longrunning.Operation;
 import com.google.protobuf.ByteString;
 import java.util.List;
@@ -168,7 +173,9 @@ public interface IBigtableTableAdminClient {
    *
    * @param request a {@link SnapshotTableRequest} object.
    * @return The long running {@link Operation} for the request.
+   * @deprecated this will be removed in a future release - please use backups API instead
    */
+  @Deprecated
   ApiFuture<Operation> snapshotTableAsync(SnapshotTableRequest request);
 
   /**
@@ -176,7 +183,9 @@ public interface IBigtableTableAdminClient {
    *
    * @param request a {@link GetSnapshotRequest} object.
    * @return The {@link Snapshot} defined by the request.
+   * @deprecated this will be removed in a future release - please use backups API instead
    */
+  @Deprecated
   ApiFuture<Snapshot> getSnapshotAsync(GetSnapshotRequest request);
 
   /**
@@ -184,14 +193,18 @@ public interface IBigtableTableAdminClient {
    *
    * @param request a {@link ListSnapshotsRequest} object.
    * @return The {@link ListSnapshotsResponse} which has the list of the snapshots in the cluster.
+   * @deprecated this will be removed in a future release - please use backups API instead
    */
+  @Deprecated
   ApiFuture<ListSnapshotsResponse> listSnapshotsAsync(ListSnapshotsRequest request);
 
   /**
    * Permanently deletes the specified snapshot.
    *
    * @param request a {@link DeleteSnapshotRequest} object.
+   * @deprecated this will be removed in a future release - please use backups API instead
    */
+  @Deprecated
   ApiFuture<Void> deleteSnapshotAsync(DeleteSnapshotRequest request);
 
   /**
@@ -199,6 +212,35 @@ public interface IBigtableTableAdminClient {
    *
    * @param request a {@link CreateTableFromSnapshotRequest} object.
    * @return The long running {@link Operation} for the request.
+   * @deprecated this will be removed in a future release - please use backups API instead
    */
+  @Deprecated
   ApiFuture<Operation> createTableFromSnapshotAsync(CreateTableFromSnapshotRequest request);
+
+  /**
+   * Creates a new backup from a table in a specific cluster.
+   *
+   * @param request a {@link CreateBackupRequest} object.
+   * @return The long running {@link Operation} for the request.
+   */
+  ApiFuture<Backup> createBackupAsync(CreateBackupRequest request);
+
+  /** Gets metadata information about the specified backup. */
+  ApiFuture<Backup> getBackupAsync(String clusterId, String backupId);
+
+  /** Update the specified backup. */
+  ApiFuture<Backup> updateBackupAsync(UpdateBackupRequest request);
+
+  /** Lists all backups associated with the specified cluster. */
+  ApiFuture<List<String>> listBackupsAsync(String clusterId);
+
+  /** Permanently deletes the specified backup. */
+  ApiFuture<Void> deleteBackupAsync(String clusterId, String backupId);
+
+  /**
+   * Creates a new table from a backup.
+   *
+   * @return The long running {@link Operation} for the request.
+   */
+  ApiFuture<RestoredTableResult> restoreTableAsync(RestoreTableRequest request);
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterName.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterName.java
index c1e860c54..5f7fdd9c8 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterName.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterName.java
@@ -86,8 +86,23 @@ public class BigtableClusterName {
    * @param snapshotId The id of the snapshot
    * @return A fully qualified snapshot name that contains the fully qualified cluster name as the
    *     parent and the snapshot name as the child.
+   * @deprecated this will be removed in a future release - please use backups API instead
    */
+  @Deprecated
   public String toSnapshotName(String snapshotId) {
     return clusterName + "/snapshots/" + snapshotId;
   }
+
+  /**
+   * Create a fully qualified backup name based on the the clusterName and the backupId. Backup name
+   * will look like:
+   * "projects/{projectId}/instances/{instanceId}/clusters/{clusterId}/backups/{backupId}".
+   *
+   * @param backupId The id of the backup
+   * @return A fully qualified backup name that contains the fully qualified cluster name as the
+   *     parent and the backup name as the child.
+   */
+  public String toBackupName(String backupId) {
+    return clusterName + "/backups/" + backupId;
+  }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceGrpcClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceGrpcClient.java
index bc0c27826..f137a13c4 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceGrpcClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceGrpcClient.java
@@ -15,8 +15,6 @@
  */
 package com.google.cloud.bigtable.grpc;
 
-import com.google.api.client.util.BackOff;
-import com.google.api.client.util.ExponentialBackOff;
 import com.google.api.core.InternalApi;
 import com.google.bigtable.admin.v2.AppProfile;
 import com.google.bigtable.admin.v2.BigtableInstanceAdminGrpc;
@@ -38,7 +36,7 @@ import com.google.bigtable.admin.v2.ListInstancesRequest;
 import com.google.bigtable.admin.v2.ListInstancesResponse;
 import com.google.bigtable.admin.v2.PartialUpdateInstanceRequest;
 import com.google.bigtable.admin.v2.UpdateAppProfileRequest;
-import com.google.common.primitives.Ints;
+import com.google.cloud.bigtable.util.OperationUtil;
 import com.google.iam.v1.GetIamPolicyRequest;
 import com.google.iam.v1.Policy;
 import com.google.iam.v1.SetIamPolicyRequest;
@@ -49,7 +47,6 @@ import com.google.longrunning.Operation;
 import com.google.longrunning.OperationsGrpc;
 import com.google.protobuf.Empty;
 import io.grpc.Channel;
-import io.grpc.protobuf.StatusProto;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -63,7 +60,7 @@ import java.util.concurrent.TimeoutException;
 public class BigtableInstanceGrpcClient implements BigtableInstanceClient {
 
   private final BigtableInstanceAdminGrpc.BigtableInstanceAdminBlockingStub instanceClient;
-  private final OperationsGrpc.OperationsBlockingStub operationsStub;
+  private final OperationUtil operationUtil;
 
   /**
    * Constructor for BigtableInstanceGrpcClient.
@@ -72,7 +69,7 @@ public class BigtableInstanceGrpcClient implements BigtableInstanceClient {
    */
   public BigtableInstanceGrpcClient(Channel channel) {
     this.instanceClient = BigtableInstanceAdminGrpc.newBlockingStub(channel);
-    operationsStub = OperationsGrpc.newBlockingStub(channel);
+    operationUtil = new OperationUtil(OperationsGrpc.newBlockingStub(channel));
   }
 
   /** {@inheritDoc} */
@@ -84,65 +81,20 @@ public class BigtableInstanceGrpcClient implements BigtableInstanceClient {
   /** {@inheritDoc} */
   @Override
   public Operation getOperation(GetOperationRequest request) {
-    return operationsStub.getOperation(request);
+    return operationUtil.getOperation(request);
   }
 
   /** {@inheritDoc} */
   @Override
   public void waitForOperation(Operation operation) throws IOException, TimeoutException {
-    waitForOperation(operation, 10, TimeUnit.MINUTES);
+    operationUtil.waitForOperation(operation, 10, TimeUnit.MINUTES);
   }
 
   /** {@inheritDoc} */
   @Override
   public void waitForOperation(Operation operation, long timeout, TimeUnit timeUnit)
       throws TimeoutException, IOException {
-    GetOperationRequest request =
-        GetOperationRequest.newBuilder().setName(operation.getName()).build();
-
-    ExponentialBackOff backOff =
-        new ExponentialBackOff.Builder()
-            .setInitialIntervalMillis(100)
-            .setMultiplier(1.3)
-            .setMaxIntervalMillis(Ints.checkedCast(TimeUnit.SECONDS.toMillis(60)))
-            .setMaxElapsedTimeMillis(Ints.checkedCast(timeUnit.toMillis(timeout)))
-            .build();
-
-    Operation currentOperationState = operation;
-
-    while (true) {
-      if (currentOperationState.getDone()) {
-        switch (currentOperationState.getResultCase()) {
-          case RESPONSE:
-            return;
-          case ERROR:
-            throw StatusProto.toStatusRuntimeException(currentOperationState.getError());
-          case RESULT_NOT_SET:
-            throw new IllegalStateException(
-                "System returned invalid response for Operation check: " + currentOperationState);
-        }
-      }
-
-      final long backOffMillis;
-      try {
-        backOffMillis = backOff.nextBackOffMillis();
-      } catch (IOException e) {
-        // Should never happen.
-        throw new RuntimeException(e);
-      }
-      if (backOffMillis == BackOff.STOP) {
-        throw new TimeoutException("Operation did not complete in time");
-      } else {
-        try {
-          Thread.sleep(backOffMillis);
-        } catch (InterruptedException e) {
-          Thread.currentThread().interrupt();
-          throw new IOException("Interrupted while waiting for operation to finish");
-        }
-      }
-
-      currentOperationState = getOperation(request);
-    }
+    operationUtil.waitForOperation(operation, timeout, timeUnit);
   }
 
   /** {@inheritDoc} */
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java
index 67869a653..91f5ae341 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClient.java
@@ -16,29 +16,40 @@
 package com.google.cloud.bigtable.grpc;
 
 import com.google.api.core.InternalExtensionOnly;
+import com.google.bigtable.admin.v2.Backup;
+import com.google.bigtable.admin.v2.CreateBackupRequest;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
 import com.google.bigtable.admin.v2.CreateTableRequest;
+import com.google.bigtable.admin.v2.DeleteBackupRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
+import com.google.bigtable.admin.v2.GetBackupRequest;
 import com.google.bigtable.admin.v2.GetSnapshotRequest;
 import com.google.bigtable.admin.v2.GetTableRequest;
+import com.google.bigtable.admin.v2.ListBackupsRequest;
+import com.google.bigtable.admin.v2.ListBackupsResponse;
 import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsResponse;
 import com.google.bigtable.admin.v2.ListTablesRequest;
 import com.google.bigtable.admin.v2.ListTablesResponse;
 import com.google.bigtable.admin.v2.ModifyColumnFamiliesRequest;
+import com.google.bigtable.admin.v2.RestoreTableRequest;
 import com.google.bigtable.admin.v2.Snapshot;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
 import com.google.bigtable.admin.v2.Table;
+import com.google.bigtable.admin.v2.UpdateBackupRequest;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.iam.v1.GetIamPolicyRequest;
 import com.google.iam.v1.Policy;
 import com.google.iam.v1.SetIamPolicyRequest;
 import com.google.iam.v1.TestIamPermissionsRequest;
 import com.google.iam.v1.TestIamPermissionsResponse;
+import com.google.longrunning.GetOperationRequest;
 import com.google.longrunning.Operation;
 import com.google.protobuf.Empty;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 /** A client for the Cloud Bigtable Table Admin API. */
@@ -203,4 +214,88 @@ public interface BigtableTableAdminClient {
   /** @deprecated Snapshots will be removed in the future */
   @Deprecated
   ListenableFuture<Operation> createTableFromSnapshotAsync(CreateTableFromSnapshotRequest request);
+
+  /**
+   * Creates a new backup from a table in a specific cluster.
+   *
+   * @param request a {@link CreateBackupRequest} object.
+   * @return The long running {@link Operation} for the request.
+   */
+  ListenableFuture<Operation> createBackupAsync(CreateBackupRequest request);
+
+  /**
+   * Gets metadata information about the specified backup.
+   *
+   * @param request a {@link GetBackupRequest} object.
+   * @return The {@link Backup} defined by the request.
+   */
+  ListenableFuture<Backup> getBackupAsync(GetBackupRequest request);
+
+  /**
+   * Updates the specified backup.
+   *
+   * @param request the request to update the Backup.
+   */
+  ListenableFuture<Backup> updateBackupAsync(UpdateBackupRequest request);
+
+  /**
+   * Lists all backups associated with the specified cluster.
+   *
+   * @param request a {@link GetBackupRequest} object.
+   * @return The {@link ListBackupsResponse} which has the list of the backups in the cluster.
+   */
+  ListenableFuture<ListBackupsResponse> listBackupsAsync(ListBackupsRequest request);
+
+  /**
+   * Permanently deletes the specified backup.
+   *
+   * @param request a {@link DeleteBackupRequest} object.
+   */
+  ListenableFuture<Empty> deleteBackupAsync(DeleteBackupRequest request);
+
+  /**
+   * Creates a new table from a backup.
+   *
+   * @param request a {@link RestoreTableRequest} object.
+   * @return The long running {@link Operation} for the request.
+   */
+  ListenableFuture<Operation> restoreTableAsync(RestoreTableRequest request);
+
+  /**
+   * Gets the latest state of a long-running operation. Clients may use this method to poll the
+   * operation result at intervals as recommended by the API service.
+   *
+   * <p>{@link #createBackupAsync(CreateBackupRequest)} and {@link
+   * #restoreTableAsync(RestoreTableRequest)} will return a {@link
+   * com.google.longrunning.Operation}. Use this method and pass in the {@link
+   * com.google.longrunning.Operation}'s name in the request to see if the Operation is done via
+   * {@link com.google.longrunning.Operation#getDone()}. The backup will not be available until that
+   * happens.
+   *
+   * @param request a {@link com.google.longrunning.GetOperationRequest} object.
+   * @return a {@link com.google.longrunning.Operation} object.
+   */
+  Operation getOperation(GetOperationRequest request);
+
+  /**
+   * Waits for the long running operation to complete by polling with exponential backoff. A default
+   * timeout of 10 minutes is used.
+   *
+   * @param operation
+   * @throws IOException
+   * @throws TimeoutException If the timeout is exceeded.
+   */
+  Operation waitForOperation(Operation operation) throws TimeoutException, IOException;
+
+  /**
+   * Waits for the long running operation to complete by polling with exponential backoff.
+   *
+   * @param operation
+   * @param timeout
+   * @param timeUnit
+   * @throws IOException
+   * @throws TimeoutException If the timeout is exceeded.
+   */
+  Operation waitForOperation(Operation operation, long timeout, TimeUnit timeUnit)
+      throws IOException, TimeoutException;
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java
index 827dc2bee..851183557 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminClientWrapper.java
@@ -18,20 +18,30 @@ package com.google.cloud.bigtable.grpc;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.InternalApi;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
+import com.google.bigtable.admin.v2.DeleteBackupRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
+import com.google.bigtable.admin.v2.GetBackupRequest;
 import com.google.bigtable.admin.v2.GetSnapshotRequest;
 import com.google.bigtable.admin.v2.GetTableRequest;
+import com.google.bigtable.admin.v2.ListBackupsRequest;
+import com.google.bigtable.admin.v2.ListBackupsResponse;
 import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsResponse;
 import com.google.bigtable.admin.v2.ListTablesRequest;
 import com.google.bigtable.admin.v2.ListTablesResponse;
 import com.google.bigtable.admin.v2.Snapshot;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
+import com.google.cloud.bigtable.admin.v2.internal.NameUtil;
+import com.google.cloud.bigtable.admin.v2.models.Backup;
+import com.google.cloud.bigtable.admin.v2.models.CreateBackupRequest;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoreTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoredTableResult;
 import com.google.cloud.bigtable.admin.v2.models.Table;
+import com.google.cloud.bigtable.admin.v2.models.UpdateBackupRequest;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.core.IBigtableTableAdminClient;
 import com.google.cloud.bigtable.util.ApiFutureUtil;
@@ -42,7 +52,14 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.longrunning.Operation;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Empty;
+import com.google.protobuf.InvalidProtocolBufferException;
+import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import javax.annotation.Nonnull;
 
 /**
@@ -58,6 +75,7 @@ public class BigtableTableAdminClientWrapper implements IBigtableTableAdminClien
 
   private final BigtableTableAdminClient delegate;
   private final BigtableInstanceName instanceName;
+  private final ExecutorService batchThreadPool;
 
   public BigtableTableAdminClientWrapper(
       @Nonnull BigtableTableAdminClient adminClient, @Nonnull BigtableOptions options) {
@@ -65,6 +83,7 @@ public class BigtableTableAdminClientWrapper implements IBigtableTableAdminClien
     Preconditions.checkNotNull(options);
     this.delegate = adminClient;
     this.instanceName = options.getInstanceName();
+    batchThreadPool = BigtableSessionSharedThreadPools.getInstance().getBatchThreadPool();
   }
 
   /** {@inheritDoc} */
@@ -306,4 +325,139 @@ public class BigtableTableAdminClientWrapper implements IBigtableTableAdminClien
   public ApiFuture<Operation> createTableFromSnapshotAsync(CreateTableFromSnapshotRequest request) {
     return ApiFutureUtil.adapt(delegate.createTableFromSnapshotAsync(request));
   }
+
+  @Override
+  public ApiFuture<Backup> createBackupAsync(CreateBackupRequest request) {
+    ListenableFuture<Operation> backupAsync =
+        delegate.createBackupAsync(
+            request.toProto(instanceName.getProjectId(), instanceName.getInstanceId()));
+    Function<Operation, Backup> function =
+        new Function<Operation, Backup>() {
+          @Override
+          public Backup apply(final Operation operation) {
+            try {
+              Future<Operation> operationFuture =
+                  batchThreadPool.submit(
+                      new Callable<Operation>() {
+                        @Override
+                        public Operation call() throws Exception {
+                          return delegate.waitForOperation(operation);
+                        }
+                      });
+
+              return Backup.fromProto(
+                  operationFuture
+                      .get()
+                      .getResponse()
+                      .unpack(com.google.bigtable.admin.v2.Backup.class));
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+              throw new RuntimeException("Interrupted while waiting for operation to finish");
+            } catch (ExecutionException | InvalidProtocolBufferException e) {
+              throw new IllegalStateException(e);
+            }
+          }
+        };
+    return ApiFutureUtil.transformAndAdapt(backupAsync, function);
+  }
+
+  @Override
+  public ApiFuture<Backup> getBackupAsync(String clusterId, String backupId) {
+    BigtableClusterName clusterName = instanceName.toClusterName(clusterId);
+    GetBackupRequest request =
+        GetBackupRequest.newBuilder().setName(clusterName.toBackupName(backupId)).build();
+    return ApiFutureUtil.transformAndAdapt(
+        delegate.getBackupAsync(request),
+        new Function<com.google.bigtable.admin.v2.Backup, Backup>() {
+          @Override
+          public Backup apply(com.google.bigtable.admin.v2.Backup backup) {
+            return Backup.fromProto(backup);
+          }
+        });
+  }
+
+  @Override
+  public ApiFuture<Backup> updateBackupAsync(UpdateBackupRequest request) {
+    return ApiFutureUtil.transformAndAdapt(
+        delegate.updateBackupAsync(
+            request.toProto(instanceName.getProjectId(), instanceName.getInstanceId())),
+        new Function<com.google.bigtable.admin.v2.Backup, Backup>() {
+          @Override
+          public Backup apply(com.google.bigtable.admin.v2.Backup backup) {
+            return Backup.fromProto(backup);
+          }
+        });
+  }
+
+  @Override
+  public ApiFuture<List<String>> listBackupsAsync(String clusterId) {
+    BigtableClusterName clusterName = instanceName.toClusterName(clusterId);
+    ListBackupsRequest request =
+        ListBackupsRequest.newBuilder().setParent(clusterName.getClusterName()).build();
+
+    // pagination is not required as page_size in the request is not set - everything will be
+    // returned
+    return ApiFutureUtil.transformAndAdapt(
+        delegate.listBackupsAsync(request),
+        new Function<ListBackupsResponse, List<String>>() {
+          @Override
+          public List<String> apply(ListBackupsResponse response) {
+            List<String> backups = new ArrayList<>();
+            for (com.google.bigtable.admin.v2.Backup backup : response.getBackupsList()) {
+              backups.add(NameUtil.extractBackupIdFromBackupName(backup.getName()));
+            }
+            return backups;
+          }
+        });
+  }
+
+  @Override
+  public ApiFuture<Void> deleteBackupAsync(String clusterId, String backupId) {
+    BigtableClusterName clusterName = instanceName.toClusterName(clusterId);
+    DeleteBackupRequest request =
+        DeleteBackupRequest.newBuilder().setName(clusterName.toBackupName(backupId)).build();
+    return ApiFutureUtil.transformAndAdapt(
+        delegate.deleteBackupAsync(request),
+        new Function<Empty, Void>() {
+          @Override
+          public Void apply(Empty empty) {
+            return null;
+          }
+        });
+  }
+
+  @Override
+  public ApiFuture<RestoredTableResult> restoreTableAsync(RestoreTableRequest request) {
+    return ApiFutureUtil.transformAndAdapt(
+        delegate.restoreTableAsync(
+            request.toProto(instanceName.getProjectId(), instanceName.getInstanceId())),
+        new Function<Operation, RestoredTableResult>() {
+          @Override
+          public RestoredTableResult apply(final Operation operation) {
+            try {
+              Future<Operation> operationFuture =
+                  batchThreadPool.submit(
+                      new Callable<Operation>() {
+                        @Override
+                        public Operation call() throws Exception {
+                          return delegate.waitForOperation(operation);
+                        }
+                      });
+
+              return new RestoredTableResult(
+                  Table.fromProto(
+                      operationFuture
+                          .get()
+                          .getResponse()
+                          .unpack(com.google.bigtable.admin.v2.Table.class)),
+                  operation
+                      .getMetadata()
+                      .unpack(com.google.bigtable.admin.v2.RestoreTableMetadata.class)
+                      .getOptimizeTableOperationName());
+            } catch (IOException | InterruptedException | ExecutionException e) {
+              throw new IllegalStateException(e);
+            }
+          }
+        });
+  }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGCJClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGCJClient.java
index 36c317446..1eb039717 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGCJClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGCJClient.java
@@ -30,9 +30,14 @@ import com.google.bigtable.admin.v2.Snapshot;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
 import com.google.cloud.bigtable.admin.v2.BaseBigtableTableAdminClient;
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminClient;
+import com.google.cloud.bigtable.admin.v2.models.Backup;
+import com.google.cloud.bigtable.admin.v2.models.CreateBackupRequest;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoreTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoredTableResult;
 import com.google.cloud.bigtable.admin.v2.models.Table;
+import com.google.cloud.bigtable.admin.v2.models.UpdateBackupRequest;
 import com.google.cloud.bigtable.core.IBigtableTableAdminClient;
 import com.google.longrunning.Operation;
 import com.google.protobuf.ByteString;
@@ -193,6 +198,41 @@ public class BigtableTableAdminGCJClient implements IBigtableTableAdminClient, A
     return baseAdminClient.createTableFromSnapshotCallable().futureCall(request);
   }
 
+  /** {@inheritDoc} */
+  @Override
+  public ApiFuture<Backup> createBackupAsync(CreateBackupRequest request) {
+    return delegate.createBackupAsync(request);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public ApiFuture<Backup> getBackupAsync(String clusterId, String backupId) {
+    return delegate.getBackupAsync(clusterId, backupId);
+  }
+
+  @Override
+  public ApiFuture<Backup> updateBackupAsync(UpdateBackupRequest request) {
+    return delegate.updateBackupAsync(request);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public ApiFuture<List<String>> listBackupsAsync(String clusterId) {
+    return delegate.listBackupsAsync(clusterId);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public ApiFuture<Void> deleteBackupAsync(String clusterId, String backupId) {
+    return delegate.deleteBackupAsync(clusterId, backupId);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public ApiFuture<RestoredTableResult> restoreTableAsync(RestoreTableRequest request) {
+    return delegate.restoreTableAsync(request);
+  }
+
   @Override
   public void close() throws Exception {
     delegate.close();
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java
index 0f91a7db4..12e726c27 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableTableAdminGrpcClient.java
@@ -21,31 +21,40 @@ import com.google.api.client.util.BackOff;
 import com.google.api.client.util.ExponentialBackOff;
 import com.google.api.core.InternalApi;
 import com.google.api.core.NanoClock;
+import com.google.bigtable.admin.v2.Backup;
 import com.google.bigtable.admin.v2.BigtableTableAdminGrpc;
 import com.google.bigtable.admin.v2.CheckConsistencyRequest;
 import com.google.bigtable.admin.v2.CheckConsistencyResponse;
+import com.google.bigtable.admin.v2.CreateBackupRequest;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
 import com.google.bigtable.admin.v2.CreateTableRequest;
+import com.google.bigtable.admin.v2.DeleteBackupRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
 import com.google.bigtable.admin.v2.GenerateConsistencyTokenRequest;
 import com.google.bigtable.admin.v2.GenerateConsistencyTokenResponse;
+import com.google.bigtable.admin.v2.GetBackupRequest;
 import com.google.bigtable.admin.v2.GetSnapshotRequest;
 import com.google.bigtable.admin.v2.GetTableRequest;
+import com.google.bigtable.admin.v2.ListBackupsRequest;
+import com.google.bigtable.admin.v2.ListBackupsResponse;
 import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsResponse;
 import com.google.bigtable.admin.v2.ListTablesRequest;
 import com.google.bigtable.admin.v2.ListTablesResponse;
 import com.google.bigtable.admin.v2.ModifyColumnFamiliesRequest;
+import com.google.bigtable.admin.v2.RestoreTableRequest;
 import com.google.bigtable.admin.v2.Snapshot;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
 import com.google.bigtable.admin.v2.Table;
+import com.google.bigtable.admin.v2.UpdateBackupRequest;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.config.RetryOptions;
 import com.google.cloud.bigtable.grpc.async.BigtableAsyncRpc;
 import com.google.cloud.bigtable.grpc.async.BigtableAsyncUtilities;
 import com.google.cloud.bigtable.grpc.async.RetryingUnaryOperation;
+import com.google.cloud.bigtable.util.OperationUtil;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Predicates;
 import com.google.common.primitives.Ints;
@@ -55,13 +64,16 @@ import com.google.iam.v1.Policy;
 import com.google.iam.v1.SetIamPolicyRequest;
 import com.google.iam.v1.TestIamPermissionsRequest;
 import com.google.iam.v1.TestIamPermissionsResponse;
+import com.google.longrunning.GetOperationRequest;
 import com.google.longrunning.Operation;
+import com.google.longrunning.OperationsGrpc;
 import com.google.protobuf.Empty;
 import io.grpc.CallOptions;
 import io.grpc.Channel;
 import io.grpc.Metadata;
 import java.io.IOException;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 /**
@@ -72,6 +84,8 @@ import java.util.concurrent.TimeoutException;
 @InternalApi("For internal usage only")
 public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
 
+  private final OperationUtil operationUtil;
+
   private final BigtableAsyncRpc<ListTablesRequest, ListTablesResponse> listTablesRpc;
   private final RetryOptions retryOptions;
   private final ScheduledExecutorService retryExecutorService;
@@ -96,6 +110,13 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
   private final BigtableAsyncRpc<CreateTableFromSnapshotRequest, Operation>
       createTableFromSnapshotRpc;
 
+  private final BigtableAsyncRpc<GetBackupRequest, Backup> getBackupRpc;
+  private final BigtableAsyncRpc<ListBackupsRequest, ListBackupsResponse> listBackupRpc;
+  private final BigtableAsyncRpc<CreateBackupRequest, Operation> createBackupRpc;
+  private final BigtableAsyncRpc<UpdateBackupRequest, Backup> updateBackupRpc;
+  private final BigtableAsyncRpc<DeleteBackupRequest, Empty> deleteBackupRpc;
+  private final BigtableAsyncRpc<RestoreTableRequest, Operation> restoreTableRpc;
+
   /**
    * Constructor for BigtableTableAdminGrpcClient.
    *
@@ -107,6 +128,8 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
       BigtableOptions bigtableOptions) {
     BigtableAsyncUtilities asyncUtilities = new BigtableAsyncUtilities.Default(channel);
 
+    operationUtil = new OperationUtil(OperationsGrpc.newBlockingStub(channel));
+
     // Read only methods.  These are always retried.
     this.listTablesRpc =
         asyncUtilities.createAsyncRpc(
@@ -116,6 +139,14 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
         asyncUtilities.createAsyncRpc(
             BigtableTableAdminGrpc.getGetTableMethod(), Predicates.<GetTableRequest>alwaysTrue());
 
+    this.listBackupRpc =
+        asyncUtilities.createAsyncRpc(
+            BigtableTableAdminGrpc.getListBackupsMethod(),
+            Predicates.<ListBackupsRequest>alwaysTrue());
+    this.getBackupRpc =
+        asyncUtilities.createAsyncRpc(
+            BigtableTableAdminGrpc.getGetBackupMethod(), Predicates.<GetBackupRequest>alwaysTrue());
+
     // Write methods. These are only retried for UNAVAILABLE or UNAUTHORIZED
     this.createTableRpc =
         asyncUtilities.createAsyncRpc(
@@ -175,6 +206,23 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
             BigtableTableAdminGrpc.getCreateTableFromSnapshotMethod(),
             Predicates.<CreateTableFromSnapshotRequest>alwaysFalse());
 
+    this.createBackupRpc =
+        asyncUtilities.createAsyncRpc(
+            BigtableTableAdminGrpc.getCreateBackupMethod(),
+            Predicates.<CreateBackupRequest>alwaysFalse());
+    this.updateBackupRpc =
+        asyncUtilities.createAsyncRpc(
+            BigtableTableAdminGrpc.getUpdateBackupMethod(),
+            Predicates.<UpdateBackupRequest>alwaysFalse());
+    this.deleteBackupRpc =
+        asyncUtilities.createAsyncRpc(
+            BigtableTableAdminGrpc.getDeleteBackupMethod(),
+            Predicates.<DeleteBackupRequest>alwaysFalse());
+    this.restoreTableRpc =
+        asyncUtilities.createAsyncRpc(
+            BigtableTableAdminGrpc.getRestoreTableMethod(),
+            Predicates.<RestoreTableRequest>alwaysFalse());
+
     this.retryOptions = bigtableOptions.getRetryOptions();
     this.retryExecutorService = retryExecutorService;
   }
@@ -302,6 +350,56 @@ public class BigtableTableAdminGrpcClient implements BigtableTableAdminClient {
         .getBlockingResult();
   }
 
+  @Override
+  public ListenableFuture<Operation> createBackupAsync(CreateBackupRequest request) {
+    return createUnaryListener(request, createBackupRpc, request.getParent()).getAsyncResult();
+  }
+
+  @Override
+  public ListenableFuture<Backup> getBackupAsync(GetBackupRequest request) {
+    return createUnaryListener(request, getBackupRpc, request.getName()).getAsyncResult();
+  }
+
+  @Override
+  public ListenableFuture<Backup> updateBackupAsync(UpdateBackupRequest request) {
+    return createUnaryListener(request, updateBackupRpc, request.getBackup().getName())
+        .getAsyncResult();
+  }
+
+  @Override
+  public ListenableFuture<ListBackupsResponse> listBackupsAsync(ListBackupsRequest request) {
+    return createUnaryListener(request, listBackupRpc, request.getParent()).getAsyncResult();
+  }
+
+  @Override
+  public ListenableFuture<Empty> deleteBackupAsync(DeleteBackupRequest request) {
+    return createUnaryListener(request, deleteBackupRpc, request.getName()).getAsyncResult();
+  }
+
+  @Override
+  public ListenableFuture<Operation> restoreTableAsync(RestoreTableRequest request) {
+    return createUnaryListener(request, restoreTableRpc, request.getParent()).getAsyncResult();
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public Operation getOperation(GetOperationRequest request) {
+    return operationUtil.getOperation(request);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public Operation waitForOperation(Operation operation) throws IOException, TimeoutException {
+    return operationUtil.waitForOperation(operation, 10, TimeUnit.MINUTES);
+  }
+
+  /** {@inheritDoc} */
+  @Override
+  public Operation waitForOperation(Operation operation, long timeout, TimeUnit timeUnit)
+      throws TimeoutException, IOException {
+    return operationUtil.waitForOperation(operation, timeout, timeUnit);
+  }
+
   @VisibleForTesting
   void waitForReplication(BigtableTableName tableName, BackOff backOff)
       throws InterruptedException, TimeoutException {
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/OperationUtil.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/OperationUtil.java
new file mode 100644
index 000000000..600892bdd
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/OperationUtil.java
@@ -0,0 +1,101 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.util;
+
+import com.google.api.client.util.BackOff;
+import com.google.api.client.util.ExponentialBackOff;
+import com.google.api.core.InternalApi;
+import com.google.common.primitives.Ints;
+import com.google.longrunning.GetOperationRequest;
+import com.google.longrunning.Operation;
+import com.google.longrunning.OperationsGrpc;
+import com.google.longrunning.OperationsGrpc.OperationsBlockingStub;
+import io.grpc.protobuf.StatusProto;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+@InternalApi("For internal usage only")
+public class OperationUtil {
+
+  private OperationsGrpc.OperationsBlockingStub operationsStub;
+
+  @InternalApi("For internal usage only")
+  public OperationUtil(OperationsBlockingStub operationsStub) {
+    this.operationsStub = operationsStub;
+  }
+
+  @InternalApi("For internal usage only")
+  public Operation getOperation(GetOperationRequest request) {
+    return operationsStub.getOperation(request);
+  }
+
+  @InternalApi("For internal usage only")
+  public void waitForOperation(Operation operation) throws IOException, TimeoutException {
+    waitForOperation(operation, 10, TimeUnit.MINUTES);
+  }
+
+  @InternalApi("For internal usage only")
+  public Operation waitForOperation(Operation operation, long timeout, TimeUnit timeUnit)
+      throws TimeoutException, IOException {
+    GetOperationRequest request =
+        GetOperationRequest.newBuilder().setName(operation.getName()).build();
+
+    ExponentialBackOff backOff =
+        new ExponentialBackOff.Builder()
+            .setInitialIntervalMillis(100)
+            .setMultiplier(1.3)
+            .setMaxIntervalMillis(Ints.checkedCast(TimeUnit.SECONDS.toMillis(60)))
+            .setMaxElapsedTimeMillis(Ints.checkedCast(timeUnit.toMillis(timeout)))
+            .build();
+
+    Operation currentOperationState = operation;
+
+    while (true) {
+      if (currentOperationState.getDone()) {
+        switch (currentOperationState.getResultCase()) {
+          case RESPONSE:
+            return currentOperationState;
+          case ERROR:
+            throw StatusProto.toStatusRuntimeException(currentOperationState.getError());
+          default:
+            throw new IllegalStateException(
+                "System returned invalid response for Operation check: " + currentOperationState);
+        }
+      }
+
+      final long backOffMillis;
+      try {
+        backOffMillis = backOff.nextBackOffMillis();
+      } catch (IOException e) {
+        // Should never happen.
+        throw new RuntimeException(e);
+      }
+      if (backOffMillis == BackOff.STOP) {
+        throw new TimeoutException("Operation did not complete in time");
+      } else {
+        try {
+          Thread.sleep(backOffMillis);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          throw new IOException("Interrupted while waiting for operation to finish");
+        }
+      }
+
+      currentOperationState = getOperation(request);
+    }
+  }
+}
diff --git a/bigtable-client-core-parent/bigtable-hbase/pom.xml b/bigtable-client-core-parent/bigtable-hbase/pom.xml
index 304cb10e3..c994c8175 100644
--- a/bigtable-client-core-parent/bigtable-hbase/pom.xml
+++ b/bigtable-client-core-parent/bigtable-hbase/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase</artifactId>
@@ -81,10 +81,6 @@ limitations under the License.
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-bigtable-admin-v2</artifactId>
     </dependency>
-    <dependency>
-      <groupId>com.google.api.grpc</groupId>
-      <artifactId>proto-google-common-protos</artifactId>
-    </dependency>
     <dependency>
       <groupId>com.google.auth</groupId>
       <artifactId>google-auth-library-credentials</artifactId>
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
index 5c196ead5..31ad40481 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
@@ -77,11 +77,17 @@ public class BigtableOptionsFactory {
       "google.bigtable.snapshot.cluster.id";
   /**
    * Constant <code>
-   * BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY="google.bigtable.snapshot.default.ttl.secs"</code>
+   * BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY="google.bigtable.snapshot.default.ttl.secs"</code> Will
+   * default to 24 hrs if not set (see BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_VALUE)
    */
   public static final String BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY =
       "google.bigtable.snapshot.default.ttl.secs";
 
+  /*
+   * Constant for default ttl for backups - 24 hours
+   */
+  public static final int BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_VALUE = 86400;
+
   /** Constant <code>CUSTOM_USER_AGENT_KEY="google.bigtable.custom.user.agent"</code> */
   public static final String CUSTOM_USER_AGENT_KEY = "google.bigtable.custom.user.agent";
 
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
index 042b8cd1c..024235351 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
@@ -22,11 +22,11 @@ import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.InternalApi;
 import com.google.api.gax.rpc.FailedPreconditionException;
-import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
-import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
-import com.google.bigtable.admin.v2.SnapshotTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.Backup;
+import com.google.cloud.bigtable.admin.v2.models.CreateBackupRequest;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoreTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.Table;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.config.Logger;
@@ -38,13 +38,12 @@ import com.google.cloud.bigtable.hbase.adapters.admin.TableAdapter;
 import com.google.cloud.bigtable.hbase.util.ModifyTableBuilder;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
-import com.google.longrunning.Operation;
 import com.google.protobuf.ByteString;
-import com.google.protobuf.Duration;
 import io.grpc.Status;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -54,7 +53,6 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.TimeoutException;
 import java.util.regex.Pattern;
 import javax.annotation.Nullable;
 import org.apache.hadoop.conf.Configuration;
@@ -72,12 +70,13 @@ import org.apache.hadoop.hbase.TableNotEnabledException;
 import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
 import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
-import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
 import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
 import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.Pair;
+import org.threeten.bp.Instant;
+import org.threeten.bp.temporal.ChronoUnit;
 
 /**
  * Abstract AbstractBigtableAdmin class.
@@ -858,84 +857,69 @@ public abstract class AbstractBigtableAdmin implements Admin {
   /**
    * Creates a snapshot from an existing table. NOTE: Cloud Bigtable has a cleanup policy
    *
-   * @param snapshotName a {@link String} object.
+   * @param snapshotId a {@link String} object.
    * @param tableName a {@link TableName} object.
    * @throws IOException if any.
    */
   @Override
-  public void snapshot(String snapshotName, TableName tableName)
+  public void snapshot(String snapshotId, TableName tableName)
       throws IOException, SnapshotCreationException, IllegalArgumentException {
 
-    Operation operation = snapshotTable(snapshotName, tableName);
-    try {
-      connection.getSession().getInstanceAdminClient().waitForOperation(operation);
-    } catch (TimeoutException e) {
-      throw new IOException("Timed out waiting for snapshot creation to finish", e);
-    }
+    snapshotTable(snapshotId, tableName);
   }
 
-  /**
-   * Creates a snapshot from an existing table. NOTE: Cloud Bigtable has a cleanup policy
-   *
-   * @param snapshotName a {@link String} object.
-   * @param tableName a {@link TableName} object.
-   * @return a {@link Operation} object.
-   * @throws IOException if any.
-   */
-  protected Operation snapshotTable(String snapshotName, TableName tableName) throws IOException {
-    Preconditions.checkNotNull(snapshotName);
-    Preconditions.checkNotNull(tableName);
-    if (snapshotName.isEmpty()) {
-      // HBase returns an empty operation instance in case snapshotName is an empty string.
-      return Operation.newBuilder().build();
-    }
+  protected Backup snapshotTable(String snapshotId, TableName tableName) throws IOException {
+    Preconditions.checkArgument(!Strings.isNullOrEmpty(snapshotId));
+    Preconditions.checkArgument(!Strings.isNullOrEmpty(tableName.getNameAsString()));
 
-    SnapshotTableRequest.Builder requestBuilder =
-        SnapshotTableRequest.newBuilder()
-            .setCluster(getSnapshotClusterName().toString())
-            .setSnapshotId(snapshotName)
-            .setName(options.getInstanceName().toTableNameStr(tableName.getNameAsString()));
+    CreateBackupRequest request =
+        CreateBackupRequest.of(getBackupClusterName().getClusterId(), snapshotId)
+            .setSourceTableId(tableName.getNameAsString());
 
     int ttlSecs =
-        configuration.getInt(BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY, -1);
-    if (ttlSecs > 0) {
-      requestBuilder.setTtl(Duration.newBuilder().setSeconds(ttlSecs).build());
-    }
+        configuration.getInt(
+            BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY,
+            BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_VALUE);
 
-    ApiFuture<Operation> future =
-        tableAdminClientWrapper.snapshotTableAsync(requestBuilder.build());
+    if (ttlSecs <= 0) {
+      throw new IllegalArgumentException(
+          BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY + " must be > 0");
+    }
+    Instant expireTime = Instant.now().plus(ttlSecs, ChronoUnit.SECONDS);
+    request.setExpireTime(expireTime);
 
-    return Futures.getChecked(future, IOException.class);
+    return Futures.getChecked(
+        tableAdminClientWrapper.createBackupAsync(request), IOException.class);
   }
 
   /**
    * This is needed for the hbase shell.
    *
-   * @param snapshotName a byte array object.
+   * @param snapshotId a byte array object.
    * @param tableName a byte array object.
    * @throws IOException if any.
    */
-  public void snapshot(byte[] snapshotName, byte[] tableName)
+  public void snapshot(byte[] snapshotId, byte[] tableName)
       throws IOException, IllegalArgumentException {
-    snapshot(snapshotName, TableName.valueOf(tableName));
+    snapshot(snapshotId, TableName.valueOf(tableName));
   }
 
   /** {@inheritDoc} */
   @Override
-  public void snapshot(byte[] snapshotName, TableName tableName)
+  public void snapshot(byte[] snapshotId, TableName tableName)
       throws IOException, IllegalArgumentException {
-    snapshot(Bytes.toString(snapshotName), tableName);
+    snapshot(Bytes.toString(snapshotId), tableName);
   }
 
   /**
    * This is needed for the hbase shell.
    *
-   * @param snapshotName a byte array object.
+   * @param snapshotId a byte array object.
    * @param tableName a byte array object.
    * @throws IOException if any.
    */
-  public void cloneSnapshot(byte[] snapshotName, byte[] tableName) throws IOException {
-    cloneSnapshot(snapshotName, TableName.valueOf(tableName));
+  public void cloneSnapshot(byte[] snapshotId, byte[] tableName) throws IOException {
+    cloneSnapshot(snapshotId, TableName.valueOf(tableName));
   }
 
   /**
@@ -950,35 +934,24 @@ public abstract class AbstractBigtableAdmin implements Admin {
   }
 
   /**
-   * @param snapshotName a {@link String} object.
+   * @param snapshotId a {@link String} object.
    * @param tableName a {@link TableName} object.
    * @throws IOException if any.
    */
   @Override
-  public void cloneSnapshot(String snapshotName, TableName tableName)
+  public void cloneSnapshot(String snapshotId, TableName tableName)
       throws IOException, TableExistsException, RestoreSnapshotException {
-    CreateTableFromSnapshotRequest request =
-        CreateTableFromSnapshotRequest.newBuilder()
-            .setParent(options.getInstanceName().toString())
-            .setTableId(tableName.getNameAsString())
-            .setSourceSnapshot(getClusterName().toSnapshotName(snapshotName))
-            .build();
-    Operation operation =
-        Futures.getChecked(
-            tableAdminClientWrapper.createTableFromSnapshotAsync(request), IOException.class);
-
-    try {
-      connection.getSession().getInstanceAdminClient().waitForOperation(operation);
-    } catch (TimeoutException e) {
-      throw new IOException("Timed out waiting for cloneSnapshot operation to finish", e);
-    }
+    RestoreTableRequest request =
+        RestoreTableRequest.of(getBackupClusterName().getClusterId(), snapshotId)
+            .setTableId(tableName.getNameAsString());
+    Futures.getChecked(tableAdminClientWrapper.restoreTableAsync(request), IOException.class);
   }
 
   protected BigtableClusterName getClusterName() throws IOException {
     return connection.getSession().getClusterName();
   }
 
-  protected BigtableClusterName getSnapshotClusterName() throws IOException {
+  protected BigtableClusterName getBackupClusterName() throws IOException {
     if (bigtableSnapshotClusterName == null) {
       try {
         bigtableSnapshotClusterName = getClusterName();
@@ -999,41 +972,26 @@ public abstract class AbstractBigtableAdmin implements Admin {
 
   /** {@inheritDoc} */
   @Override
-  public void deleteSnapshot(String snapshotName) throws IOException {
-    Preconditions.checkNotNull(snapshotName);
-    if (snapshotName.isEmpty()) {
+  public void deleteSnapshot(String snapshotId) throws IOException {
+    Preconditions.checkNotNull(snapshotId);
+    if (snapshotId.isEmpty()) {
       return;
     }
 
-    String btSnapshotName = getClusterName().toSnapshotName(snapshotName);
-    DeleteSnapshotRequest request =
-        DeleteSnapshotRequest.newBuilder().setName(btSnapshotName).build();
-
-    Futures.getUnchecked(tableAdminClientWrapper.deleteSnapshotAsync(request));
+    Futures.getChecked(
+        tableAdminClientWrapper.deleteBackupAsync(
+            getBackupClusterName().getClusterId(), snapshotId),
+        IOException.class);
   }
 
-  /**
-   * {@inheritDoc}
-   *
-   * <p>The snapshots will be deleted serially and the first failure will prevent the deletion of
-   * the remaining snapshots.
-   */
   @Override
   public void deleteSnapshots(String regex) throws IOException {
-    deleteSnapshots(Pattern.compile(regex));
+    throw new UnsupportedOperationException("use deleteSnapshot instead");
   }
 
-  /**
-   * {@inheritDoc}
-   *
-   * <p>The snapshots will be deleted serially and the first failure will prevent the deletion of
-   * the remaining snapshots.
-   */
   @Override
   public void deleteSnapshots(Pattern pattern) throws IOException {
-    for (SnapshotDescription snapshotDescription : listSnapshots(pattern)) {
-      deleteSnapshot(snapshotDescription.getName());
-    }
+    throw new UnsupportedOperationException("use deleteSnapshot instead");
   }
 
   /** {@inheritDoc} */
@@ -1052,10 +1010,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
   @Override
   public void deleteTableSnapshots(Pattern tableNamePattern, Pattern snapshotNamePattern)
       throws IOException {
-    for (SnapshotDescription snapshotDescription :
-        listTableSnapshots(tableNamePattern, snapshotNamePattern)) {
-      deleteSnapshot(snapshotDescription.getName());
-    }
+    throw new UnsupportedOperationException("Unsupported - please use deleteSnapshots");
   }
 
   // ------------- Unsupported snapshot methods.
diff --git a/bigtable-client-core-parent/pom.xml b/bigtable-client-core-parent/pom.xml
index 91619663f..0be69bd64 100644
--- a/bigtable-client-core-parent/pom.xml
+++ b/bigtable-client-core-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-client-core-parent</artifactId>
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
index ba0e62ff8..037dded10 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
+++ b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-dataflow-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml b/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
index cbe9748c4..6a24c4dde 100755
--- a/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
+++ b/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
@@ -16,7 +16,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-dataflow-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-beam</artifactId>
diff --git a/bigtable-dataflow-parent/pom.xml b/bigtable-dataflow-parent/pom.xml
index dc84fbaa9..3062775fb 100644
--- a/bigtable-dataflow-parent/pom.xml
+++ b/bigtable-dataflow-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-dataflow-parent</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
index e76c62fcb..883c02520 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
@@ -3,7 +3,7 @@
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
index 1273870fa..12344f51d 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
index 8aae11c05..66cb493d1 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-1.x-mapreduce</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
index 581fe9178..9ffaf71ea 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
index b3de7b3c9..3f6214349 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-1.x</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/main/java/com/google/cloud/bigtable/hbase1_x/BigtableAdmin.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/main/java/com/google/cloud/bigtable/hbase1_x/BigtableAdmin.java
index e9d58439f..953cbb25c 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/main/java/com/google/cloud/bigtable/hbase1_x/BigtableAdmin.java
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/main/java/com/google/cloud/bigtable/hbase1_x/BigtableAdmin.java
@@ -16,11 +16,6 @@
 package com.google.cloud.bigtable.hbase1_x;
 
 import com.google.api.core.InternalApi;
-import com.google.bigtable.admin.v2.ListSnapshotsRequest;
-import com.google.bigtable.admin.v2.ListSnapshotsResponse;
-import com.google.bigtable.admin.v2.Snapshot;
-import com.google.cloud.bigtable.grpc.BigtableSnapshotName;
-import com.google.cloud.bigtable.grpc.BigtableTableName;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.Futures;
 import java.io.IOException;
@@ -110,9 +105,9 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
    */
   @Override
   public void snapshot(
-      String snapshotName, TableName tableName, HBaseProtos.SnapshotDescription.Type type)
+      String snapshotId, TableName tableName, HBaseProtos.SnapshotDescription.Type type)
       throws IOException, SnapshotCreationException, IllegalArgumentException {
-    snapshot(snapshotName, tableName);
+    snapshot(snapshotId, tableName);
   }
 
   /** {@inheritDoc} */
@@ -131,32 +126,31 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   @Override
   public void restoreSnapshot(String s, boolean b, boolean b1)
-      throws IOException, RestoreSnapshotException {}
+      throws IOException, RestoreSnapshotException {
+    throw new UnsupportedOperationException("restoreSnapshot"); // TODO
+  }
 
   @Override
   public void cloneSnapshot(String s, TableName tableName, boolean b)
-      throws IOException, TableExistsException, RestoreSnapshotException {}
+      throws IOException, TableExistsException, RestoreSnapshotException {
+    if (!b) {
+      this.cloneSnapshot(s, tableName);
+    }
+    throw new UnsupportedOperationException("cloneSnapshot"); // TODO
+  }
 
   /** {@inheritDoc} */
   @Override
   public List<HBaseProtos.SnapshotDescription> listSnapshots() throws IOException {
-    ListSnapshotsRequest request =
-        ListSnapshotsRequest.newBuilder().setParent(getSnapshotClusterName().toString()).build();
-
-    ListSnapshotsResponse snapshotList =
-        Futures.getChecked(tableAdminClientWrapper.listSnapshotsAsync(request), IOException.class);
+    List<String> backups =
+        Futures.getChecked(
+            tableAdminClientWrapper.listBackupsAsync(getBackupClusterName().getClusterId()),
+            IOException.class);
 
     List<HBaseProtos.SnapshotDescription> response = new ArrayList<>();
 
-    for (Snapshot snapshot : snapshotList.getSnapshotsList()) {
-      BigtableSnapshotName snapshotName = new BigtableSnapshotName(snapshot.getName());
-      BigtableTableName tableName = new BigtableTableName(snapshot.getSourceTable().getName());
-      response.add(
-          HBaseProtos.SnapshotDescription.newBuilder()
-              .setName(snapshotName.getSnapshotId())
-              .setTable(tableName.getTableId())
-              .setCreationTime(TimeUnit.SECONDS.toMillis(snapshot.getCreateTime().getSeconds()))
-              .build());
+    for (String snapshot : backups) {
+      response.add(HBaseProtos.SnapshotDescription.newBuilder().setName(snapshot).build());
     }
     return response;
   }
@@ -186,23 +180,13 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
   @Override
   public List<SnapshotDescription> listTableSnapshots(
       String tableNameRegex, String snapshotNameRegex) throws IOException {
-    return listTableSnapshots(Pattern.compile(tableNameRegex), Pattern.compile(snapshotNameRegex));
+    throw new UnsupportedOperationException("Unsupported - please use listSnapshots");
   }
 
   @Override
   public List<SnapshotDescription> listTableSnapshots(
       Pattern tableNamePattern, Pattern snapshotNamePattern) throws IOException {
-    if (tableNamePattern == null || tableNamePattern.matcher("").matches()) {
-      return ImmutableList.of();
-    }
-
-    List<SnapshotDescription> response = new ArrayList<>();
-    for (SnapshotDescription snapshotDescription : listSnapshots(snapshotNamePattern)) {
-      if (tableNamePattern.matcher(snapshotDescription.getTable()).matches()) {
-        response.add(snapshotDescription);
-      }
-    }
-    return response;
+    throw new UnsupportedOperationException("Unsupported - please use listSnapshots");
   }
 
   @Override
diff --git a/bigtable-hbase-1.x-parent/pom.xml b/bigtable-hbase-1.x-parent/pom.xml
index 5d3e70adc..03b7bd457 100644
--- a/bigtable-hbase-1.x-parent/pom.xml
+++ b/bigtable-hbase-1.x-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-1.x-parent</artifactId>
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
index 1dd62399b..83a6c5390 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
index 2228afffa..351795a9e 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
index c3a939c39..d6c0c144f 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <properties>
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
index 7bc5c5511..c0c2b6736 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
@@ -16,9 +16,6 @@
 package com.google.cloud.bigtable.hbase2_x;
 
 import com.google.api.core.InternalApi;
-import com.google.bigtable.admin.v2.ListSnapshotsRequest;
-import com.google.bigtable.admin.v2.ListSnapshotsResponse;
-import com.google.bigtable.admin.v2.Snapshot;
 import com.google.cloud.bigtable.hbase.util.ModifyTableBuilder;
 import com.google.cloud.bigtable.hbase2_x.adapters.admin.TableAdapter2x;
 import com.google.common.collect.ImmutableList;
@@ -142,20 +139,22 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   @Override
   public List<SnapshotDescription> listSnapshots() throws IOException {
-    ListSnapshotsRequest request =
-        ListSnapshotsRequest.newBuilder().setParent(getSnapshotClusterName().toString()).build();
-
-    ListSnapshotsResponse snapshotList =
-        Futures.getChecked(tableAdminClientWrapper.listSnapshotsAsync(request), IOException.class);
+    List<String> backups =
+        Futures.getChecked(
+            tableAdminClientWrapper.listBackupsAsync(getBackupClusterName().getClusterId()),
+            IOException.class);
     List<SnapshotDescription> response = new ArrayList<>();
-    for (Snapshot snapshot : snapshotList.getSnapshotsList()) {
-      response.add(
-          new SnapshotDescription(
-              snapshot.getName(), TableName.valueOf(snapshot.getSourceTable().getName())));
+    for (String backup : backups) {
+      response.add(new SnapshotDescription(backup));
     }
     return response;
   }
 
+  @Override
+  public void deleteSnapshots(Pattern pattern) throws IOException {
+    throw new UnsupportedOperationException("use deleteSnapshot instead");
+  }
+
   /**
    * {@inheritDoc}
    *
@@ -226,9 +225,9 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   /** {@inheritDoc} */
   @Override
-  public void snapshot(String snapshotName, TableName tableName, SnapshotType arg2)
+  public void snapshot(String snapshotId, TableName tableName, SnapshotType arg2)
       throws IOException, SnapshotCreationException, IllegalArgumentException {
-    snapshot(snapshotName, tableName);
+    snapshot(snapshotId, tableName);
   }
 
   /** {@inheritDoc} */
@@ -303,25 +302,15 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
   }
 
   @Override
-  public List<SnapshotDescription> listTableSnapshots(String tableName, String snapshotName)
+  public List<SnapshotDescription> listTableSnapshots(String tableName, String snapshotId)
       throws IOException {
-    return listTableSnapshots(Pattern.compile(tableName), Pattern.compile(snapshotName));
+    throw new UnsupportedOperationException("Unsupported - please use listSnapshots");
   }
 
   @Override
   public List<SnapshotDescription> listTableSnapshots(Pattern tableName, Pattern snapshotName)
       throws IOException {
-    if (tableName == null || tableName.matcher("").matches()) {
-      return ImmutableList.of();
-    }
-
-    List<SnapshotDescription> response = new ArrayList<>();
-    for (SnapshotDescription snapshot : listSnapshots(snapshotName)) {
-      if (tableName.matcher(snapshot.getTableNameAsString()).matches()) {
-        response.add(snapshot);
-      }
-    }
-    return response;
+    throw new UnsupportedOperationException("Unsupported - please use listSnapshots");
   }
 
   @Override
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
index a6f2cf7b8..f0040b002 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncAdmin.java
@@ -19,12 +19,10 @@ import static com.google.cloud.bigtable.hbase2_x.FutureUtils.failedFuture;
 import static com.google.cloud.bigtable.hbase2_x.FutureUtils.toCompletableFuture;
 
 import com.google.api.core.InternalApi;
-import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
-import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
-import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.Snapshot;
-import com.google.bigtable.admin.v2.SnapshotTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.CreateBackupRequest;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoreTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.Table;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.config.Logger;
@@ -35,6 +33,7 @@ import com.google.cloud.bigtable.hbase.BigtableOptionsFactory;
 import com.google.cloud.bigtable.hbase.util.ModifyTableBuilder;
 import com.google.cloud.bigtable.hbase2_x.adapters.admin.TableAdapter2x;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import io.grpc.Status;
 import java.io.IOException;
@@ -85,8 +84,9 @@ import org.apache.hadoop.hbase.security.access.GetUserPermissionsRequest;
 import org.apache.hadoop.hbase.security.access.Permission;
 import org.apache.hadoop.hbase.security.access.UserPermission;
 import org.apache.hadoop.hbase.shaded.com.google.protobuf.RpcChannel;
-import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.threeten.bp.Instant;
+import org.threeten.bp.temporal.ChronoUnit;
 
 /**
  * Bigtable implementation of {@link AsyncAdmin}
@@ -103,6 +103,7 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
   private final TableAdapter2x tableAdapter2x;
   private final CommonConnection asyncConnection;
   private BigtableClusterName bigtableSnapshotClusterName;
+  private final int ttlSeconds;
 
   public BigtableAsyncAdmin(CommonConnection asyncConnection) throws IOException {
     LOG.debug("Creating BigtableAsyncAdmin");
@@ -119,6 +120,15 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     if (clusterId != null) {
       bigtableSnapshotClusterName = bigtableInstanceName.toClusterName(clusterId);
     }
+    this.ttlSeconds =
+        configuration.getInt(
+            BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY,
+            BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_VALUE);
+
+    if (this.ttlSeconds <= 0) {
+      throw new IllegalArgumentException(
+          BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY + " must be > 0");
+    }
   }
 
   /** {@inheritDoc} */
@@ -334,25 +344,20 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
   }
 
   @Override
-  public CompletableFuture<Void> deleteSnapshot(String snapshotName) {
-    return CompletableFuture.supplyAsync(
-            () -> {
-              try {
-                return DeleteSnapshotRequest.newBuilder()
-                    .setName(getClusterName().toSnapshotName(snapshotName))
-                    .build();
-              } catch (IOException e) {
-                throw new CompletionException(e);
-              }
-            })
-        .thenAccept(bigtableTableAdminClient::deleteSnapshotAsync);
+  public CompletableFuture<Void> deleteSnapshot(String snapshotId) {
+    try {
+      return toCompletableFuture(
+          bigtableTableAdminClient.deleteBackupAsync(
+              getBackupClusterName().getClusterId(), snapshotId));
+    } catch (IOException e) {
+      throw new CompletionException(e);
+    }
   }
 
   @Override
   public CompletableFuture<Void> deleteTableSnapshots(
       Pattern tableNamePattern, Pattern snapshotNamePattern) {
-    return listTableSnapshots(tableNamePattern, snapshotNamePattern)
-        .thenApply(deleteSnapshotsFunc());
+    throw new UnsupportedOperationException("Unsupported - please use deleteSnapshots");
   }
 
   //  ******************* START COLUMN FAMILY MODIFICATION  ************************
@@ -402,73 +407,27 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
 
   //  ******************* END COLUMN FAMILY MODIFICATION  ************************
 
-  /**
-   * Restore the specified snapshot on the original table.
-   *
-   * <p>{@inheritDoc}
-   */
+  /** In place table restorations are not supported. Please use cloneSnapshot */
   @Override
   public CompletableFuture<Void> restoreSnapshot(String snapshotName) {
-    return restoreSnapshot(snapshotName, false);
+    throw new UnsupportedOperationException(
+        "In place table restorations are not supported. Please use cloneSnapshot");
   }
 
-  /**
-   * Restore the specified snapshot on the original table.
-   *
-   * <p>{@inheritDoc}
-   */
+  /** In place table restorations are not supported. Please use cloneSnapshot */
   @Override
   public CompletableFuture<Void> restoreSnapshot(
       String snapshotName, boolean takeFailSafeSnapshot) {
-    CompletableFuture<Void> future = new CompletableFuture<>();
-    listSnapshots(Pattern.compile(snapshotName))
-        .whenComplete(
-            (snapshotDescriptions, err) -> {
-              if (err != null) {
-                future.completeExceptionally(err);
-                return;
-              }
-              final TableName tableName = snapshotExists(snapshotName, snapshotDescriptions);
-              if (tableName == null) {
-                future.completeExceptionally(
-                    new RestoreSnapshotException(
-                        "Unable to find the table name for snapshot=" + snapshotName));
-                return;
-              }
-              tableExists(tableName)
-                  .whenComplete(
-                      (exists, err2) -> {
-                        if (err2 != null) {
-                          future.completeExceptionally(err2);
-                        } else if (!exists) {
-                          // if table does not exist, then just clone snapshot into new table.
-                          completeConditionalOnFuture(
-                              future, cloneSnapshot(snapshotName, tableName));
-                        } else {
-                          isTableDisabled(tableName)
-                              .whenComplete(
-                                  (disabled, err4) -> {
-                                    if (err4 != null) {
-                                      future.completeExceptionally(err4);
-                                    } else if (!disabled) {
-                                      future.completeExceptionally(
-                                          new TableNotDisabledException(tableName));
-                                    } else {
-                                      completeConditionalOnFuture(
-                                          future,
-                                          restoreSnapshot(snapshotName, takeFailSafeSnapshot));
-                                    }
-                                  });
-                        }
-                      });
-            });
-    return future;
+    throw new UnsupportedOperationException(
+        "In place table restorations are not supported. Please use cloneSnapshot");
   }
 
+  /** Same table restorations are not supported. Please use cloneSnapshot */
   @Override
   public CompletableFuture<Void> restoreSnapshot(
       String snapshotName, boolean takeFailSafeSnapshot, boolean restoreAcl) {
-    throw new UnsupportedOperationException("restoreSnapshot");
+    throw new UnsupportedOperationException(
+        "In place table restorations are not supported. Please use cloneSnapshot");
   }
 
   /**
@@ -515,61 +474,63 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
         bigtableTableAdminClient.dropAllRowsAsync(tableName.getNameAsString()));
   }
 
-  private Function<List<SnapshotDescription>, Void> deleteSnapshotsFunc() {
-    return snapshots -> {
-      snapshots.stream().forEach(desc -> deleteSnapshot(desc.getName()));
-      return null;
-    };
-  }
-
+  /*
+   * deleteSnapshots is not supported due to backup write quota limits - please use deleteSnapshot
+   */
   @Override
   public CompletableFuture<Void> deleteSnapshots() {
-    return listSnapshots().thenApply(deleteSnapshotsFunc());
+    throw new UnsupportedOperationException("use deleteSnapshot instead");
   }
 
+  /*
+   * deleteSnapshots is not supported due to backup write quota limits - please use deleteSnapshot
+   */
   @Override
   public CompletableFuture<Void> deleteSnapshots(Pattern pattern) {
-    return listSnapshots(pattern).thenApply(deleteSnapshotsFunc());
+    throw new UnsupportedOperationException("use deleteSnapshot instead");
   }
 
+  /*
+   * deleteSnapshots is not supported due to backup write quota limits - please use deleteSnapshot
+   */
   @Override
   public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern) {
-    return listSnapshots(tableNamePattern).thenApply(deleteSnapshotsFunc());
+    throw new UnsupportedOperationException("use deleteSnapshot instead");
   }
 
   @Override
-  public CompletableFuture<Void> snapshot(String snapshotName, TableName tableName) {
-    return CompletableFuture.supplyAsync(
-            () -> {
-              try {
-                return SnapshotTableRequest.newBuilder()
-                    .setCluster(getSnapshotClusterName().toString())
-                    .setSnapshotId(snapshotName)
-                    .setName(bigtableInstanceName.toTableNameStr(tableName.getNameAsString()))
-                    .build();
-              } catch (IOException e) {
-                throw new CompletionException(e);
-              }
-            })
-        .thenAccept(c -> toCompletableFuture(bigtableTableAdminClient.snapshotTableAsync(c)));
+  public CompletableFuture<Void> snapshot(String snapshotId, TableName tableName) {
+    Instant expireTime = Instant.now().plus(ttlSeconds, ChronoUnit.SECONDS);
+    if (Strings.isNullOrEmpty(snapshotId)) {
+      throw new IllegalArgumentException("Snapshot name cannot be null");
+    }
+    if (Strings.isNullOrEmpty(tableName.getNameAsString())) {
+      throw new IllegalArgumentException("Table name cannot be null");
+    }
+
+    try {
+      return toCompletableFuture(
+              bigtableTableAdminClient.createBackupAsync(
+                  CreateBackupRequest.of(getBackupClusterName().getClusterId(), snapshotId)
+                      .setExpireTime(expireTime)
+                      .setSourceTableId(tableName.getNameAsString())))
+          .thenAccept(backup -> {});
+    } catch (IOException e) {
+      throw new CompletionException(e);
+    }
   }
 
   @Override
-  public CompletableFuture<Void> cloneSnapshot(String snapshotName, TableName tableName) {
-    return CompletableFuture.supplyAsync(
-            () -> {
-              try {
-                return CreateTableFromSnapshotRequest.newBuilder()
-                    .setParent(bigtableInstanceName.toString())
-                    .setTableId(tableName.getNameAsString())
-                    .setSourceSnapshot(getClusterName().toSnapshotName(snapshotName))
-                    .build();
-              } catch (IOException e) {
-                throw new CompletionException(e);
-              }
-            })
-        .thenAccept(
-            c -> toCompletableFuture(bigtableTableAdminClient.createTableFromSnapshotAsync(c)));
+  public CompletableFuture<Void> cloneSnapshot(String snapshotId, TableName tableName) {
+    try {
+      return toCompletableFuture(
+              bigtableTableAdminClient.restoreTableAsync(
+                  RestoreTableRequest.of(getBackupClusterName().getClusterId(), snapshotId)
+                      .setTableId(tableName.getNameAsString())))
+          .thenAccept(backup -> {});
+    } catch (IOException e) {
+      throw new CompletionException(e);
+    }
   }
 
   @Override
@@ -583,20 +544,18 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     return CompletableFuture.supplyAsync(
             () -> {
               try {
-                return ListSnapshotsRequest.newBuilder()
-                    .setParent(getSnapshotClusterName().toString())
-                    .build();
+                return getBackupClusterName().getClusterId();
               } catch (IOException e) {
                 throw new CompletionException(e);
               }
             })
         .thenCompose(
-            request ->
-                toCompletableFuture(bigtableTableAdminClient.listSnapshotsAsync(request))
+            c ->
+                toCompletableFuture(bigtableTableAdminClient.listBackupsAsync(c))
                     .thenApply(
                         r ->
-                            r.getSnapshotsList().stream()
-                                .map(BigtableAsyncAdmin::toSnapshotDescription)
+                            r.stream()
+                                .map(b -> new SnapshotDescription(b))
                                 .collect(Collectors.toList())));
   }
 
@@ -612,14 +571,7 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
   @Override
   public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(
       Pattern tableNamePattern, Pattern snapshotPattern) {
-    Preconditions.checkNotNull(tableNamePattern);
-    if (tableNamePattern.matcher("").matches()) {
-      return CompletableFuture.completedFuture(ImmutableList.of());
-    }
-
-    return listSnapshots(snapshotPattern)
-        .thenApply(
-            r -> filter(r, d -> tableNamePattern.matcher(d.getTableNameAsString()).matches()));
+    throw new UnsupportedOperationException("Unsupported - please use listSnapshots");
   }
 
   private static SnapshotDescription toSnapshotDescription(Snapshot snapshot) {
@@ -708,14 +660,7 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
 
   @Override
   public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern) {
-    return listSnapshots()
-        .thenApply(
-            r ->
-                filter(
-                    r,
-                    d ->
-                        tableNamePattern == null
-                            || tableNamePattern.matcher(d.getTableNameAsString()).matches()));
+    throw new UnsupportedOperationException("Unsupported - please use listSnapshots");
   }
 
   @Override
@@ -746,7 +691,7 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
         });
   }
 
-  private BigtableClusterName getSnapshotClusterName() throws IOException {
+  private BigtableClusterName getBackupClusterName() throws IOException {
     if (bigtableSnapshotClusterName == null) {
       try {
         bigtableSnapshotClusterName = getClusterName();
diff --git a/bigtable-hbase-2.x-parent/pom.xml b/bigtable-hbase-2.x-parent/pom.xml
index be9c3bfef..f9cee7bb5 100644
--- a/bigtable-hbase-2.x-parent/pom.xml
+++ b/bigtable-hbase-2.x-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-2.x-parent</artifactId>
diff --git a/pom.xml b/pom.xml
index d8efca1be..148e83843 100644
--- a/pom.xml
+++ b/pom.xml
@@ -19,7 +19,7 @@ limitations under the License.
 
   <groupId>com.google.cloud.bigtable</groupId>
   <artifactId>bigtable-client-parent</artifactId>
-  <version>1.14.2-SNAPSHOT</version>
+  <version>1.15.1-SNAPSHOT</version>
   <packaging>pom</packaging>
   <name>${project.groupId}:${project.artifactId}</name>
   <url>https://cloud.google.com/bigtable/</url>
@@ -50,7 +50,7 @@ limitations under the License.
     <compileSource.1.8>1.8</compileSource.1.8>
 
     <!-- core dependency versions -->
-    <bigtable.version>1.11.0</bigtable.version>
+    <bigtable.version>1.14.0</bigtable.version>
     <dropwizard.metrics.version>3.2.6</dropwizard.metrics.version>
     <slf4j.version>1.7.30</slf4j.version>
     <commons-logging.version>1.2</commons-logging.version>


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/FakeBigtableAdminServiceImpl.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/FakeBigtableAdminServiceImpl.java
index 19e56bb95..8d391a713 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/FakeBigtableAdminServiceImpl.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/FakeBigtableAdminServiceImpl.java
@@ -15,26 +15,37 @@
  */
 package com.google.cloud.bigtable.grpc;
 
+import com.google.bigtable.admin.v2.Backup;
 import com.google.bigtable.admin.v2.BigtableTableAdminGrpc.BigtableTableAdminImplBase;
 import com.google.bigtable.admin.v2.ColumnFamily;
+import com.google.bigtable.admin.v2.CreateBackupMetadata;
+import com.google.bigtable.admin.v2.CreateBackupRequest;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
 import com.google.bigtable.admin.v2.CreateTableRequest;
+import com.google.bigtable.admin.v2.DeleteBackupRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
+import com.google.bigtable.admin.v2.GetBackupRequest;
 import com.google.bigtable.admin.v2.GetSnapshotRequest;
 import com.google.bigtable.admin.v2.GetTableRequest;
+import com.google.bigtable.admin.v2.ListBackupsRequest;
+import com.google.bigtable.admin.v2.ListBackupsResponse;
 import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsResponse;
 import com.google.bigtable.admin.v2.ListTablesRequest;
 import com.google.bigtable.admin.v2.ListTablesResponse;
 import com.google.bigtable.admin.v2.ModifyColumnFamiliesRequest;
 import com.google.bigtable.admin.v2.ModifyColumnFamiliesRequest.Modification;
+import com.google.bigtable.admin.v2.RestoreTableMetadata;
+import com.google.bigtable.admin.v2.RestoreTableRequest;
 import com.google.bigtable.admin.v2.Snapshot;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
 import com.google.bigtable.admin.v2.Table;
+import com.google.bigtable.admin.v2.UpdateBackupRequest;
 import com.google.cloud.bigtable.data.v2.internal.NameUtil;
 import com.google.longrunning.Operation;
+import com.google.protobuf.Any;
 import com.google.protobuf.Empty;
 import com.google.protobuf.GeneratedMessageV3;
 import io.grpc.stub.StreamObserver;
@@ -151,6 +162,96 @@ public class FakeBigtableAdminServiceImpl extends BigtableTableAdminImplBase {
     responseObserver.onCompleted();
   }
 
+  @Override
+  public void createBackup(
+      CreateBackupRequest request, StreamObserver<Operation> responseObserver) {
+    responseObserver.onNext(
+        Operation.newBuilder()
+            .setMetadata(
+                Any.pack(CreateBackupMetadata.newBuilder().setName(request.getBackupId()).build()))
+            .setResponse(
+                Any.pack(
+                    com.google.bigtable.admin.v2.Backup.newBuilder()
+                        .setName(
+                            "projects/fake-project-id/instances/fake-instance-id/clusters/fake-cluster-id/backups/fake-backup-id")
+                        .setSourceTable(
+                            "projects/fake-project-id/instances/fake-instance-id/tables/fake-table-id")
+                        .build()))
+            .setDone(true)
+            .build());
+    responseObserver.onCompleted();
+  }
+
+  @Override
+  public void getBackup(GetBackupRequest request, StreamObserver<Backup> responseObserver) {
+    responseObserver.onNext(
+        Backup.newBuilder()
+            .setName(
+                "projects/fake-project-id/instances/fake-instance-id/clusters/fake-cluster-id/backups/fake-backup-id")
+            .setSourceTable(
+                "projects/fake-project-id/instances/fake-instance-id/tables/fake-table-id")
+            .build());
+    responseObserver.onCompleted();
+  }
+
+  @Override
+  public void updateBackup(UpdateBackupRequest request, StreamObserver<Backup> responseObserver) {
+    responseObserver.onNext(
+        Backup.newBuilder()
+            .setName(
+                "projects/fake-project-id/instances/fake-instance-id/clusters/fake-cluster-id/backups/fake-backup-id")
+            .setSourceTable(
+                "projects/fake-project-id/instances/fake-instance-id/tables/fake-table-id")
+            .setExpireTime(request.getBackup().getExpireTime())
+            .build());
+    responseObserver.onCompleted();
+  }
+
+  @Override
+  public void listBackups(
+      ListBackupsRequest request, StreamObserver<ListBackupsResponse> responseObserver) {
+    ListBackupsResponse listBackups =
+        ListBackupsResponse.newBuilder()
+            .addBackups(
+                Backup.newBuilder()
+                    .setName(
+                        "projects/fake-project-id/instances/fake-instance-id/clusters/fake-cluster-id/backups/fake-backup-1")
+                    .build())
+            .addBackups(
+                Backup.newBuilder()
+                    .setName(
+                        "projects/fake-project-id/instances/fake-instance-id/clusters/fake-cluster-id/backups/fake-backup-2")
+                    .build())
+            .build();
+    responseObserver.onNext(listBackups);
+    responseObserver.onCompleted();
+  }
+
+  @Override
+  public void deleteBackup(DeleteBackupRequest request, StreamObserver<Empty> responseObserver) {
+    requests.add(request);
+    responseObserver.onNext(Empty.newBuilder().build());
+    responseObserver.onCompleted();
+  }
+
+  @Override
+  public void restoreTable(
+      RestoreTableRequest request, StreamObserver<Operation> responseObserver) {
+    responseObserver.onNext(
+        Operation.newBuilder()
+            .setMetadata(
+                Any.pack(RestoreTableMetadata.newBuilder().setName(request.getTableId()).build()))
+            .setResponse(
+                Any.pack(
+                    com.google.bigtable.admin.v2.Table.newBuilder()
+                        .setName(
+                            "projects/fake-project-id/instances/fake-instance-id/tables/fake-table-id")
+                        .build()))
+            .setDone(true)
+            .build());
+    responseObserver.onCompleted();
+  }
+
   public List<GeneratedMessageV3> getRequests() {
     return new ArrayList<>(requests);
   }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableClusterName.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableClusterName.java
index 33c65f64f..c9112f25a 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableClusterName.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableClusterName.java
@@ -38,4 +38,12 @@ public class TestBigtableClusterName {
     Assert.assertEquals(
         clusterName + "/snapshots/snp", new BigtableClusterName(clusterName).toSnapshotName("snp"));
   }
+
+  @Test
+  public void createBackupName() throws Exception {
+    String clusterName = "projects/proj/instances/inst/clusters/cluster1";
+    Assert.assertEquals(
+        clusterName + "/backups/backup",
+        new BigtableClusterName(clusterName).toBackupName("backup"));
+  }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java
index e398c57e0..2cd30b821 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminClientWrapper.java
@@ -23,41 +23,60 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.bigtable.admin.v2.ColumnFamily;
+import com.google.bigtable.admin.v2.CreateBackupMetadata;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
+import com.google.bigtable.admin.v2.DeleteBackupRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
+import com.google.bigtable.admin.v2.GetBackupRequest;
 import com.google.bigtable.admin.v2.GetSnapshotRequest;
 import com.google.bigtable.admin.v2.GetTableRequest;
+import com.google.bigtable.admin.v2.ListBackupsRequest;
+import com.google.bigtable.admin.v2.ListBackupsResponse;
 import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsResponse;
 import com.google.bigtable.admin.v2.ListTablesRequest;
 import com.google.bigtable.admin.v2.ListTablesResponse;
+import com.google.bigtable.admin.v2.RestoreTableMetadata;
 import com.google.bigtable.admin.v2.Snapshot;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
+import com.google.cloud.bigtable.admin.v2.internal.NameUtil;
+import com.google.cloud.bigtable.admin.v2.models.Backup;
+import com.google.cloud.bigtable.admin.v2.models.CreateBackupRequest;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.GCRules;
 import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoreTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoredTableResult;
 import com.google.cloud.bigtable.admin.v2.models.Table;
+import com.google.cloud.bigtable.admin.v2.models.UpdateBackupRequest;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.Futures;
 import com.google.longrunning.Operation;
+import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Empty;
+import com.google.protobuf.Timestamp;
+import com.google.protobuf.util.Timestamps;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.mockito.Mockito;
+import org.threeten.bp.Instant;
 
 @RunWith(JUnit4.class)
 public class TestBigtableTableAdminClientWrapper {
   private static final String PROJECT_ID = "projectId";
   private static final String INSTANCE_ID = "instanceId";
+  private static final String CLUSTER_ID = "clusterId";
   private static final String TABLE_ID = "tableId";
+  private static final String BACKUP_ID = "backupId";
   private static final String APP_PROFILE_ID = "appProfileId";
   private static final String COLUMN_FAMILY = "myColumnFamily";
   private static final String ROW_KEY_PREFIX = "row-key-val";
@@ -76,6 +95,9 @@ public class TestBigtableTableAdminClientWrapper {
       new BigtableInstanceName(PROJECT_ID, INSTANCE_ID);
   private static final String TABLE_NAME = INSTANCE_NAME.toTableNameStr(TABLE_ID);
 
+  private static final BigtableClusterName CLUSTER_NAME = INSTANCE_NAME.toClusterName(CLUSTER_ID);
+  private static final String BACKUP_NAME = CLUSTER_NAME.toBackupName(BACKUP_ID);
+
   private BigtableTableAdminClient mockAdminClient;
 
   private BigtableTableAdminClientWrapper clientWrapper;
@@ -334,6 +356,116 @@ public class TestBigtableTableAdminClientWrapper {
     verify(mockAdminClient).createTableFromSnapshotAsync(request);
   }
 
+  @Test
+  public void testCreateBackupAsync() throws Exception {
+    CreateBackupRequest request =
+        CreateBackupRequest.of(CLUSTER_ID, BACKUP_ID).setSourceTableId(TABLE_NAME);
+    com.google.bigtable.admin.v2.CreateBackupRequest requestProto =
+        request.toProto(PROJECT_ID, INSTANCE_ID);
+    Operation op =
+        Operation.newBuilder()
+            .setMetadata(Any.pack(CreateBackupMetadata.getDefaultInstance()))
+            .setResponse(
+                Any.pack(
+                    com.google.bigtable.admin.v2.Backup.newBuilder()
+                        .setName(BACKUP_NAME)
+                        .setSourceTable(TABLE_NAME)
+                        .build()))
+            .build();
+    when(mockAdminClient.waitForOperation(op)).thenReturn(op);
+    when(mockAdminClient.createBackupAsync(requestProto)).thenReturn(Futures.immediateFuture(op));
+    Future<Backup> actualResponse = clientWrapper.createBackupAsync(request);
+    assertEquals(BACKUP_ID, actualResponse.get().getId());
+    verify(mockAdminClient).createBackupAsync(requestProto);
+  }
+
+  @Test
+  public void testGetBackupAsync() throws Exception {
+    GetBackupRequest request = GetBackupRequest.newBuilder().setName(BACKUP_NAME).build();
+    com.google.bigtable.admin.v2.Backup response =
+        com.google.bigtable.admin.v2.Backup.newBuilder()
+            .setName(BACKUP_NAME)
+            .setSourceTable(TABLE_NAME)
+            .build();
+    when(mockAdminClient.getBackupAsync(request)).thenReturn(Futures.immediateFuture(response));
+    Future<Backup> actualResponse = clientWrapper.getBackupAsync(CLUSTER_ID, BACKUP_ID);
+    assertEquals(BACKUP_ID, actualResponse.get().getId());
+    assertEquals(TABLE_ID, actualResponse.get().getSourceTableId());
+    verify(mockAdminClient).getBackupAsync(request);
+  }
+
+  @Test
+  public void testUpdateBackupAsync() throws ExecutionException, InterruptedException {
+    // Setup
+    Timestamp expireTime = Timestamp.newBuilder().setSeconds(123456789).build();
+    long sizeBytes = 12345L;
+    UpdateBackupRequest req = UpdateBackupRequest.of(CLUSTER_ID, BACKUP_ID);
+    com.google.bigtable.admin.v2.Backup response =
+        com.google.bigtable.admin.v2.Backup.newBuilder()
+            .setName(NameUtil.formatBackupName(PROJECT_ID, INSTANCE_ID, CLUSTER_ID, BACKUP_ID))
+            .setSourceTable(NameUtil.formatTableName(PROJECT_ID, INSTANCE_ID, TABLE_ID))
+            .setExpireTime(expireTime)
+            .setSizeBytes(sizeBytes)
+            .build();
+    Mockito.when(mockAdminClient.updateBackupAsync(req.toProto(PROJECT_ID, INSTANCE_ID)))
+        .thenReturn(Futures.immediateFuture(response));
+
+    Backup actualResult = clientWrapper.updateBackupAsync(req).get();
+
+    assertEquals(actualResult.getId(), BACKUP_ID);
+    assertEquals(actualResult.getSourceTableId(), TABLE_ID);
+    assertEquals(
+        actualResult.getExpireTime(), Instant.ofEpochMilli(Timestamps.toMillis(expireTime)));
+    assertEquals(actualResult.getSizeBytes(), sizeBytes);
+  }
+
+  @Test
+  public void testListBackupsAsync() throws Exception {
+    ListBackupsRequest request =
+        ListBackupsRequest.newBuilder().setParent(CLUSTER_NAME.toString()).build();
+    ListBackupsResponse response =
+        ListBackupsResponse.newBuilder()
+            .addBackups(
+                com.google.bigtable.admin.v2.Backup.newBuilder()
+                    .setName(BACKUP_NAME)
+                    .setSourceTable(TABLE_NAME)
+                    .build())
+            .build();
+    when(mockAdminClient.listBackupsAsync(request)).thenReturn(Futures.immediateFuture(response));
+    Future<List<String>> actualResponse = clientWrapper.listBackupsAsync(CLUSTER_ID);
+    assertEquals(response.getBackupsCount(), actualResponse.get().size());
+    verify(mockAdminClient).listBackupsAsync(request);
+  }
+
+  @Test
+  public void testDeleteBackupAsync() {
+    DeleteBackupRequest request = DeleteBackupRequest.newBuilder().setName(BACKUP_NAME).build();
+    when(mockAdminClient.deleteBackupAsync(request))
+        .thenReturn(Futures.immediateFuture(Empty.getDefaultInstance()));
+    clientWrapper.deleteBackupAsync(CLUSTER_ID, BACKUP_ID);
+    verify(mockAdminClient).deleteBackupAsync(request);
+  }
+
+  @Test
+  public void testRestoreTableAsync() throws Exception {
+    RestoreTableRequest request =
+        RestoreTableRequest.of(CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID);
+    com.google.bigtable.admin.v2.RestoreTableRequest requestProto =
+        request.toProto(PROJECT_ID, INSTANCE_ID);
+    Operation op =
+        Operation.newBuilder()
+            .setMetadata(Any.pack(RestoreTableMetadata.newBuilder().setName(BACKUP_NAME).build()))
+            .setResponse(
+                Any.pack(
+                    com.google.bigtable.admin.v2.Table.newBuilder().setName(TABLE_NAME).build()))
+            .build();
+    when(mockAdminClient.waitForOperation(op)).thenReturn(op);
+    when(mockAdminClient.restoreTableAsync(requestProto)).thenReturn(Futures.immediateFuture(op));
+    Future<RestoredTableResult> actualResponse = clientWrapper.restoreTableAsync(request);
+    assertEquals(TABLE_ID, actualResponse.get().getTable().getId());
+    verify(mockAdminClient).restoreTableAsync(requestProto);
+  }
+
   private static com.google.bigtable.admin.v2.Table createTableData() {
     GCRules.GCRule gcRule = GCRULES.maxVersions(1);
     ColumnFamily columnFamily = ColumnFamily.newBuilder().setGcRule(gcRule.toProto()).build();
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGCJClient.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGCJClient.java
index ab1e111f7..5420d6eb9 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGCJClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGCJClient.java
@@ -18,7 +18,9 @@ package com.google.cloud.bigtable.grpc;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import com.google.bigtable.admin.v2.BackupName;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
+import com.google.bigtable.admin.v2.DeleteBackupRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.DeleteTableRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
@@ -32,10 +34,15 @@ import com.google.cloud.bigtable.admin.v2.BaseBigtableTableAdminSettings;
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminClient;
 import com.google.cloud.bigtable.admin.v2.BigtableTableAdminSettings;
 import com.google.cloud.bigtable.admin.v2.internal.NameUtil;
+import com.google.cloud.bigtable.admin.v2.models.Backup;
 import com.google.cloud.bigtable.admin.v2.models.ColumnFamily;
+import com.google.cloud.bigtable.admin.v2.models.CreateBackupRequest;
 import com.google.cloud.bigtable.admin.v2.models.CreateTableRequest;
 import com.google.cloud.bigtable.admin.v2.models.ModifyColumnFamiliesRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoreTableRequest;
+import com.google.cloud.bigtable.admin.v2.models.RestoredTableResult;
 import com.google.cloud.bigtable.admin.v2.models.Table;
+import com.google.cloud.bigtable.admin.v2.models.UpdateBackupRequest;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.config.BigtableVeneerSettingsFactory;
 import com.google.cloud.bigtable.config.CredentialOptions;
@@ -51,13 +58,16 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
+import org.threeten.bp.Instant;
 
 @RunWith(JUnit4.class)
 public class TestBigtableTableAdminGCJClient {
 
   private static final String PROJECT_ID = "fake-project-id";
   private static final String INSTANCE_ID = "fake-instance-id";
-  private static final String TABLE_ID = "fake-Table-id";
+  private static final String TABLE_ID = "fake-table-id";
+  private static final String CLUSTER_ID = "fake-cluster-id";
+  private static final String BACKUP_ID = "fake-backup-id";
   private static final String TEST_TABLE_ID_1 = "test-table-1";
   private static final String TEST_TABLE_ID_2 = "test-table-2";
   private static final String TEST_TABLE_ID_3 = "test-table-3";
@@ -251,6 +261,51 @@ public class TestBigtableTableAdminGCJClient {
     assertTrue(actualResponse.get().getDone());
   }
 
+  @Test
+  public void testCreateAndUpdateBackupAsync() throws Exception {
+    CreateBackupRequest request =
+        CreateBackupRequest.of(CLUSTER_ID, BACKUP_ID).setSourceTableId(tableName);
+    Future<Backup> actualResponse = adminGCJClient.createBackupAsync(request);
+    assertEquals(BACKUP_ID, actualResponse.get().getId());
+    assertEquals(Instant.EPOCH, actualResponse.get().getExpireTime());
+
+    Instant expireTime = Instant.ofEpochMilli(12345L);
+    Future<Backup> updateResponse =
+        adminGCJClient.updateBackupAsync(
+            UpdateBackupRequest.of(CLUSTER_ID, BACKUP_ID).setExpireTime(expireTime));
+    assertEquals(BACKUP_ID, updateResponse.get().getId());
+    assertEquals(expireTime, updateResponse.get().getExpireTime());
+  }
+
+  @Test
+  public void testGetBackupAsync() throws Exception {
+    Future<Backup> actualResponse = adminGCJClient.getBackupAsync(CLUSTER_ID, BACKUP_ID);
+    assertEquals(BACKUP_ID, actualResponse.get().getId());
+  }
+
+  @Test
+  public void testListBackupsAsync() throws Exception {
+    Future<List<String>> actualResponse = adminGCJClient.listBackupsAsync(CLUSTER_ID);
+    assertEquals("fake-backup-1", actualResponse.get().get(0));
+    assertEquals("fake-backup-2", actualResponse.get().get(1));
+  }
+
+  @Test
+  public void testDeleteBackupAsync() throws Exception {
+    String backupName = BackupName.format(PROJECT_ID, INSTANCE_ID, CLUSTER_ID, BACKUP_ID);
+    adminGCJClient.deleteBackupAsync(CLUSTER_ID, BACKUP_ID).get();
+    DeleteBackupRequest receivedReq = (DeleteBackupRequest) serviceImpl.getRequests().get(0);
+    assertEquals(backupName, receivedReq.getName());
+  }
+
+  @Test
+  public void testRestoreTableAsync() throws Exception {
+    RestoreTableRequest request =
+        RestoreTableRequest.of(CLUSTER_ID, BACKUP_ID).setTableId(TABLE_ID);
+    Future<RestoredTableResult> actualResponse = adminGCJClient.restoreTableAsync(request);
+    assertEquals(TABLE_ID, actualResponse.get().getTable().getId());
+  }
+
   @After
   public void tearDown() throws Exception {
     if (adminClientV2 != null) {
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGrpcClient.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGrpcClient.java
index d523a6471..2195991cf 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGrpcClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableTableAdminGrpcClient.java
@@ -24,16 +24,21 @@ import static org.mockito.Mockito.when;
 
 import com.google.bigtable.admin.v2.CheckConsistencyRequest;
 import com.google.bigtable.admin.v2.CheckConsistencyResponse;
+import com.google.bigtable.admin.v2.CreateBackupRequest;
 import com.google.bigtable.admin.v2.CreateTableFromSnapshotRequest;
 import com.google.bigtable.admin.v2.CreateTableRequest;
+import com.google.bigtable.admin.v2.DeleteBackupRequest;
 import com.google.bigtable.admin.v2.DeleteSnapshotRequest;
 import com.google.bigtable.admin.v2.DropRowRangeRequest;
 import com.google.bigtable.admin.v2.GenerateConsistencyTokenRequest;
 import com.google.bigtable.admin.v2.GenerateConsistencyTokenResponse;
+import com.google.bigtable.admin.v2.GetBackupRequest;
 import com.google.bigtable.admin.v2.GetSnapshotRequest;
 import com.google.bigtable.admin.v2.GetTableRequest;
+import com.google.bigtable.admin.v2.ListBackupsRequest;
 import com.google.bigtable.admin.v2.ListSnapshotsRequest;
 import com.google.bigtable.admin.v2.ModifyColumnFamiliesRequest;
+import com.google.bigtable.admin.v2.RestoreTableRequest;
 import com.google.bigtable.admin.v2.SnapshotTableRequest;
 import com.google.bigtable.admin.v2.Table;
 import com.google.cloud.bigtable.config.BigtableOptions;
@@ -223,6 +228,54 @@ public class TestBigtableTableAdminGrpcClient {
     verifyRequestCalled(request);
   }
 
+  @Test
+  public void testBackupTable() throws Exception {
+    CreateBackupRequest request =
+        CreateBackupRequest.newBuilder()
+            .setBackupId("backupId")
+            .setBackup(
+                com.google.bigtable.admin.v2.Backup.newBuilder()
+                    .setName("backupId")
+                    .setSourceTable(TABLE_NAME))
+            .build();
+    complete();
+    defaultClient.createBackupAsync(request).get(1, TimeUnit.SECONDS);
+    verifyRequestCalled(request);
+  }
+
+  @Test
+  public void testGetBackup() throws Exception {
+    GetBackupRequest request = GetBackupRequest.newBuilder().setName(TABLE_NAME).build();
+    complete();
+    defaultClient.getBackupAsync(request).get(1, TimeUnit.SECONDS);
+    verifyRequestCalled(request);
+  }
+
+  @Test
+  public void testListBackups() throws Exception {
+    ListBackupsRequest request = ListBackupsRequest.newBuilder().setParent(TABLE_NAME).build();
+    complete();
+    defaultClient.listBackupsAsync(request).get(1, TimeUnit.SECONDS);
+    verifyRequestCalled(request);
+  }
+
+  @Test
+  public void testDeleteBackup() throws Exception {
+    DeleteBackupRequest request = DeleteBackupRequest.newBuilder().setName(TABLE_NAME).build();
+    complete();
+    defaultClient.deleteBackupAsync(request).get(1, TimeUnit.SECONDS);
+    verifyRequestCalled(request);
+  }
+
+  @Test
+  public void testCreateTableFromBackup() throws Exception {
+    RestoreTableRequest request =
+        RestoreTableRequest.newBuilder().setParent(INSTANCE_NAME.getInstanceName()).build();
+    complete();
+    defaultClient.restoreTableAsync(request).get(1, TimeUnit.SECONDS);
+    verifyRequestCalled(request);
+  }
+
   private void complete() {
     setResponse("");
   }
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
index 6127b1735..b2985468b 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
@@ -19,7 +19,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-integration-tests-common</artifactId>
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestSnapshot.java b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestSnapshot.java
index 6e307b136..ebcfb809d 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestSnapshot.java
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/AbstractTestSnapshot.java
@@ -17,28 +17,17 @@
 package com.google.cloud.bigtable.hbase;
 
 import static com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule.COLUMN_FAMILY;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
-import java.util.UUID;
+import java.util.concurrent.ExecutionException;
 import java.util.regex.Pattern;
 import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.Admin;
-import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
-import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.After;
 import org.junit.Assert;
@@ -54,45 +43,30 @@ import org.junit.runners.JUnit4;
 @SuppressWarnings("deprecation")
 @RunWith(JUnit4.class)
 public abstract class AbstractTestSnapshot extends AbstractTest {
+  protected final Logger LOG = new Logger(getClass());
 
-  protected final byte[] QUALIFIER = dataHelper.randomData("TestSnapshots");
+  protected static String prefix;
+
+  protected static final int[] BACKOFF_DURATION = {2, 4, 8, 16, 32, 64, 128, 256, 512, 1024};
+
+  protected static final String TEST_BACKUP_SUFFIX = "backup-it";
+
+  protected final TableName tableName = sharedTestEnv.newTestTableName();
+  protected Map<String, Long> values;
 
-  protected TableName tableName = sharedTestEnv.newTestTableName();
-  protected final TableName anotherTableName = sharedTestEnv.newTestTableName();
-  // The maximum size of a table id or snapshot id is 50. newTestTableName().size() can approach 50.
-  // Make sure that the snapshot name and cloned table are within the 50 character limit
-  protected final String snapshotName = tableName.getNameAsString().substring(40) + "_snp";
-  protected final String anotherSnapshotName =
-      anotherTableName.getNameAsString().substring(40) + "_snp";
-  protected final TableName clonedTableName =
-      TableName.valueOf(tableName.getNameAsString().substring(40) + "_clone");
+  protected final byte[] QUALIFIER = dataHelper.randomData("TestSnapshots");
   protected HTableDescriptor descriptor;
 
   @After
   public void cleanup() {
-    if (sharedTestEnv.isBigtable() && !enableTestForBigtable()) {
-      return;
-    }
-    try (Admin admin = getConnection().getAdmin()) {
-      delete(admin, tableName);
-      delete(admin, anotherTableName);
-      delete(admin, clonedTableName);
-      for (SnapshotDescription snapDesc : admin.listSnapshots(snapshotName + ".*")) {
-        deleteSnapshot(snapDesc.getName());
-      }
-      for (SnapshotDescription snapDesc : admin.listSnapshots(anotherSnapshotName + ".*")) {
-        deleteSnapshot(snapDesc.getName());
-      }
+    try {
+      delete(tableName);
     } catch (Exception e) {
       e.printStackTrace();
     }
   }
 
-  protected boolean enableTestForBigtable() {
-    return false;
-  }
-
-  private void delete(Admin admin, TableName tableName) throws IOException {
+  private void delete(TableName tableName) throws IOException {
     if (tableExists(tableName)) {
       disableTable(tableName);
       deleteTable(tableName);
@@ -100,192 +74,81 @@ public abstract class AbstractTestSnapshot extends AbstractTest {
   }
 
   @Test
-  public void testSnapshot() throws IOException {
-    if (sharedTestEnv.isBigtable() && !enableTestForBigtable()) {
-      return;
-    }
-    try (Admin admin = getConnection().getAdmin()) {
-      createTable(tableName);
-
-      Map<String, Long> values = createAndPopulateTable(tableName);
-      checkSnapshotCount(admin, 0);
+  public void testSnapshot() throws IOException, InterruptedException, ExecutionException {
+    String snapshotName = generateId("test-snapshot");
+    try {
       snapshot(snapshotName, tableName);
-      checkSnapshotCount(admin, 1);
-      cloneSnapshot(snapshotName, clonedTableName);
-      validateClone(values);
-      checkSnapshotCount(admin, 1);
+      Assert.assertEquals(1, listSnapshotsSize(snapshotName));
+    } finally {
       deleteSnapshot(snapshotName);
-      checkSnapshotCount(admin, 0);
+      Assert.assertEquals(0, listSnapshotsSize(snapshotName));
     }
   }
 
   @Test
-  public void testListSnapshots() throws IOException {
-    if (sharedTestEnv.isBigtable() && !enableTestForBigtable()) {
-      return;
-    }
-    final Pattern allSnapshots = Pattern.compile(snapshotName + ".*");
-    createTable(tableName);
-    Map<String, Long> values = createAndPopulateTable(tableName);
-    Assert.assertEquals(0, listSnapshotsSize(allSnapshots));
-    snapshot(snapshotName, tableName);
-    Assert.assertEquals(1, listSnapshotsSize(snapshotName));
-    deleteSnapshot(snapshotName);
-    Assert.assertEquals(0, listSnapshotsSize(snapshotName));
-    snapshot(snapshotName + 1, tableName);
-    snapshot(snapshotName + 2, tableName);
-    Assert.assertEquals(2, listSnapshotsSize(allSnapshots));
-    Assert.assertEquals(1, listSnapshotsSize(Pattern.compile(snapshotName + 1)));
-    Assert.assertEquals(1, listSnapshotsSize(Pattern.compile(snapshotName + 2)));
-    deleteSnapshots(allSnapshots);
-    Assert.assertEquals(0, listSnapshotsSize(allSnapshots));
-  }
+  public void testListAndDeleteSnapshots()
+      throws IOException, InterruptedException, ExecutionException {
+    String snapshotName = generateId("list-1");
+    String snapshotName2 = generateId("list-2");
+    Pattern allSnapshots = Pattern.compile(prefix + ".*" + "list" + ".*");
 
-  @Test
-  public void testTableSnapshots() throws IOException {
-    if (sharedTestEnv.isBigtable() && !enableTestForBigtable()) {
-      return;
-    }
-    final Pattern matchAll = Pattern.compile(".*");
-    createTable(tableName);
-    createTable(anotherTableName);
-    createAndPopulateTable(tableName);
-    createAndPopulateTable(anotherTableName);
-    Assert.assertEquals(
-        0, listTableSnapshotsSize(Pattern.compile(tableName.getNameAsString()), matchAll));
-    Assert.assertEquals(
-        0, listTableSnapshotsSize(Pattern.compile(anotherTableName.getNameAsString()), matchAll));
-    snapshot(snapshotName, tableName);
-    Assert.assertEquals(
-        1, listTableSnapshotsSize(Pattern.compile(tableName.getNameAsString()), matchAll));
-    Assert.assertEquals(
-        0, listTableSnapshotsSize(Pattern.compile(anotherTableName.getNameAsString()), matchAll));
-    snapshot(anotherSnapshotName, anotherTableName);
-    Assert.assertEquals(
-        1, listTableSnapshotsSize(Pattern.compile(tableName.getNameAsString()), matchAll));
-    Assert.assertEquals(
-        1, listTableSnapshotsSize(Pattern.compile(anotherTableName.getNameAsString()), matchAll));
-    deleteSnapshot(snapshotName);
-    Assert.assertEquals(0, listTableSnapshotsSize(tableName.getNameAsString(), snapshotName));
-    Assert.assertEquals(
-        0, listTableSnapshotsSize(anotherTableName.getNameAsString(), snapshotName));
-    Assert.assertEquals(
-        1, listTableSnapshotsSize(anotherTableName.getNameAsString(), anotherSnapshotName));
-    deleteSnapshot(anotherSnapshotName);
-    Assert.assertEquals(0, listTableSnapshotsSize(tableName.getNameAsString(), snapshotName));
-    Assert.assertEquals(
-        0, listTableSnapshotsSize(anotherTableName.getNameAsString(), snapshotName));
-    Assert.assertEquals(
-        0, listTableSnapshotsSize(anotherTableName.getNameAsString(), anotherSnapshotName));
-  }
+    try {
+      snapshot(snapshotName, tableName);
+      Assert.assertEquals(1, listSnapshotsSize(Pattern.compile(snapshotName)));
 
-  @Test
-  public void testListTableSnapshotWithSingleArgs() throws Exception {
-    if (sharedTestEnv.isBigtable() && !enableTestForBigtable()) {
-      return;
+      snapshot(snapshotName2, tableName);
+      Assert.assertEquals(2, listSnapshotsSize(allSnapshots));
+
+      Assert.assertEquals(1, listSnapshotsSize(Pattern.compile(snapshotName2)));
+    } finally {
+      deleteSnapshot(snapshotName);
+      deleteSnapshot(snapshotName2);
+      Assert.assertEquals(0, listSnapshotsSize(allSnapshots));
     }
-    sharedTestEnv.createTable(tableName);
-    snapshot(snapshotName, tableName);
-    Assert.assertEquals(1, listTableSnapshotsSize(Pattern.compile(tableName.getNameAsString())));
-    snapshot(snapshotName + 1, tableName);
-    Assert.assertEquals(2, listTableSnapshotsSize(Pattern.compile(tableName.getNameAsString())));
-    deleteSnapshot(snapshotName);
-    Assert.assertEquals(1, listTableSnapshotsSize(Pattern.compile(tableName.getNameAsString())));
   }
 
   @Test
-  public void testSnapshotTableWithNullAndEmptyString() {
-    Exception actualError = null;
-    try {
-      snapshot("test-snapshot", null);
-      Assert.fail("listTableSnapshots with null tableName should fail");
-    } catch (Exception ex) {
-      actualError = ex;
-    }
-    assertNotNull(actualError);
-    actualError = null;
+  public void testCloneSnapshot() throws IOException, InterruptedException {
+    String cloneSnapshotName = generateId("clone-test");
+    snapshot(cloneSnapshotName, tableName);
 
-    try {
-      snapshot(null, tableName);
-    } catch (Exception ex) {
-      actualError = ex;
-    }
-    assertNotNull(actualError);
-  }
+    // Wait 2 minutes so that the RestoreTable API will trigger an optimize restored
+    // table operation.
+    Thread.sleep(120 * 1000);
+    Assert.assertEquals(1, listSnapshotsSize(cloneSnapshotName));
 
-  @Test
-  public void testListSnapshotsWithNullAndEmptyString() throws IOException {
-    Exception actualError = null;
+    TableName clonedTableName = TableName.valueOf(tableName.getNameAsString() + "_clne");
     try {
-      listSnapshotsSize((String) null);
-    } catch (Exception ex) {
-      actualError = ex;
-    }
-    assertNotNull(actualError);
-    assertTrue(actualError instanceof NullPointerException);
+      cloneSnapshot(cloneSnapshotName, clonedTableName);
 
-    assertEquals(0, listSnapshotsSize((Pattern) null));
+      // give table time to create
+      Thread.sleep(60 * 1000);
 
-    assertEquals(0, listSnapshotsSize(""));
+      validateClone(clonedTableName, values);
+    } finally {
+      deleteSnapshot(cloneSnapshotName);
+      delete(clonedTableName);
+    }
   }
 
   @Test
-  public void testListTableSnapshotsWithNullAndEmptyString() throws IOException {
+  public void testListSnapshotsWithNullAndEmptyString() throws IOException {
     Exception actualError = null;
     try {
-      listTableSnapshotsSize(null, "");
-      Assert.fail("listTableSnapshots with null tableName should fail");
+      listSnapshotsSize((String) null);
     } catch (Exception ex) {
       actualError = ex;
     }
-    assertNotNull(actualError);
-    assertTrue(actualError instanceof NullPointerException);
+    Assert.assertNotNull(actualError);
+    Assert.assertTrue(actualError instanceof NullPointerException);
 
-    assertEquals(0, listTableSnapshotsSize((Pattern) null, null));
+    Assert.assertEquals(0, listSnapshotsSize((Pattern) null));
 
-    assertEquals(0, listTableSnapshotsSize("", ""));
+    Assert.assertEquals(0, listSnapshotsSize(""));
   }
 
-  @Test
-  public void testDeleteSnapshotWithEmptyString() throws Exception {
-    // No snapshot matches with null or empty string hence no exception should be thrown
-    deleteSnapshots(null);
-
-    deleteSnapshots(Pattern.compile(""));
-
-    deleteTableSnapshots(null, null);
-
-    deleteTableSnapshots(Pattern.compile(""), Pattern.compile(""));
-  }
-
-  private void checkSnapshotCount(Admin admin, int count) throws IOException {
-    Assert.assertEquals(count, listSnapshotsSize(snapshotName));
-  }
-
-  /**
-   * Create a test table, and add a small number of rows to the table.
-   *
-   * @return A Map of the data that was added to the original table.
-   * @throws IOException
-   */
-  protected Map<String, Long> createAndPopulateTable(TableName tableName) throws IOException {
-    Map<String, Long> values = new HashMap<>();
-    try (Table table = getConnection().getTable(tableName)) {
-      values.clear();
-      List<Put> puts = new ArrayList<>();
-      for (long i = 0; i < 10; i++) {
-        final UUID rowKey = UUID.randomUUID();
-        byte[] row = Bytes.toBytes(rowKey.toString());
-        values.put(rowKey.toString(), i);
-        puts.add(new Put(row).addColumn(COLUMN_FAMILY, QUALIFIER, Bytes.toBytes(i)));
-      }
-      table.put(puts);
-    }
-    return values;
-  }
-
-  protected void validateClone(Map<String, Long> values) throws IOException {
-    try (Table clonedTable = getConnection().getTable(clonedTableName);
+  protected void validateClone(TableName tableName, Map<String, Long> values) throws IOException {
+    try (Table clonedTable = getConnection().getTable(tableName);
         ResultScanner scanner = clonedTable.getScanner(new Scan())) {
       for (Result result : scanner) {
         String row = Bytes.toString(result.getRow());
@@ -298,35 +161,31 @@ public abstract class AbstractTestSnapshot extends AbstractTest {
     Assert.assertTrue("There were missing keys.", values.isEmpty());
   }
 
-  protected abstract void createTable(TableName tableName) throws IOException;
+  protected abstract void snapshot(String snapshotName, TableName tableName)
+      throws IOException, InterruptedException;
 
-  protected abstract void snapshot(String snapshotName, TableName tableName) throws IOException;
+  protected abstract Map<String, Long> createAndPopulateTable(TableName tableName)
+      throws IOException, ExecutionException, InterruptedException;
+
+  protected abstract int listSnapshotsSize() throws IOException;
 
   protected abstract int listSnapshotsSize(String regEx) throws IOException;
 
   protected abstract int listSnapshotsSize(Pattern pattern) throws IOException;
 
-  protected abstract void deleteSnapshot(String snapshotName) throws IOException;
-
-  protected abstract void deleteTableSnapshots(Pattern tableName, Pattern snapshotName)
-      throws IOException;
+  protected abstract void deleteSnapshot(String snapshotName)
+      throws IOException, InterruptedException;
 
   protected abstract boolean tableExists(final TableName tableName) throws IOException;
 
   protected abstract void disableTable(final TableName tableName) throws IOException;
 
   protected abstract void cloneSnapshot(final String snapshotName, final TableName tableName)
-      throws IOException, TableExistsException, RestoreSnapshotException;
-
-  protected abstract int listTableSnapshotsSize(String tableNameRegex, String snapshotNameRegex)
       throws IOException;
 
-  protected abstract int listTableSnapshotsSize(
-      Pattern tableNamePattern, Pattern snapshotNamePattern) throws IOException;
-
-  protected abstract void deleteSnapshots(final Pattern pattern) throws IOException;
-
   protected abstract void deleteTable(final TableName tableName) throws IOException;
 
-  protected abstract int listTableSnapshotsSize(Pattern tableNamePattern) throws Exception;
+  protected static String generateId(String name) {
+    return prefix + "-" + name + "-" + TEST_BACKUP_SUFFIX;
+  }
 }
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
index bab544389..592e71f2f 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-1.x-integration-tests</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java
index bf614a0fd..7942bff63 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java
@@ -16,56 +16,87 @@
 package com.google.cloud.bigtable.hbase;
 
 import static com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule.COLUMN_FAMILY;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.MatcherAssert.assertThat;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Before;
 import org.junit.Test;
 
 public class TestSnapshots extends AbstractTestSnapshot {
 
-  @Test
-  public void testSnapshotWithSingleParam() throws IOException {
-    Exception actualError = null;
+  @Before
+  public void setUp() throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      try {
-        admin.snapshot(null);
-      } catch (Exception e) {
-        actualError = e;
+      // Setup a prefix to avoid collisions between concurrent test runs
+      prefix = String.format("020%d", System.currentTimeMillis());
+
+      // clean up stale backups
+      String stalePrefix =
+          String.format("020%d", System.currentTimeMillis() - TimeUnit.HOURS.toMillis(2));
+
+      for (SnapshotDescription snapshotDescription : admin.listSnapshots()) {
+        int i = snapshotDescription.getName().lastIndexOf("/");
+        String backupId = snapshotDescription.getName().substring(i + 1);
+        if (backupId.endsWith(TEST_BACKUP_SUFFIX) && stalePrefix.compareTo(backupId) > 0) {
+          LOG.info("Deleting old snapshot: " + backupId);
+          admin.deleteSnapshot(backupId);
+        }
       }
-      assertNotNull(actualError);
-      assertTrue(actualError instanceof NullPointerException);
-      actualError = null;
-
-      try {
-        admin.snapshot("", tableName, null);
-      } catch (Exception e) {
-        actualError = e;
+      values = createAndPopulateTable(tableName);
+    }
+  }
+
+  @Test
+  public void listSnapshots() throws IOException, InterruptedException {
+    String snapshot1 = generateId("snapshot-1");
+    snapshot(snapshot1, tableName);
+
+    String snapshot2 = generateId("snapshot-2");
+    snapshot(snapshot2, tableName);
+
+    try (Admin admin = getConnection().getAdmin()) {
+      List<SnapshotDescription> snapshotDescriptions = admin.listSnapshots();
+      List<String> descStrings = new ArrayList<>();
+      for (SnapshotDescription snapshotDescription : snapshotDescriptions) {
+        descStrings.add(snapshotDescription.getName());
       }
-      assertNotNull(actualError);
+      assertThat(descStrings, hasItems(snapshot1, snapshot2));
+    } finally {
+      deleteSnapshot(snapshot1);
+      deleteSnapshot(snapshot2);
     }
   }
 
   @Override
-  protected void createTable(TableName tableName) throws IOException {
+  protected void snapshot(String snapshotName, TableName tableName)
+      throws IOException, InterruptedException {
     try (Admin admin = getConnection().getAdmin()) {
-      HTableDescriptor descriptor = new HTableDescriptor(tableName);
-      descriptor.addFamily(new HColumnDescriptor(COLUMN_FAMILY));
-      admin.createTable(descriptor);
+      admin.snapshot(snapshotName, tableName);
     }
   }
 
   @Override
-  protected void snapshot(String snapshotName, TableName tableName) throws IOException {
+  protected int listSnapshotsSize() throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      admin.snapshot(snapshotName, tableName);
+      return admin.listSnapshots().size();
     }
   }
 
@@ -77,7 +108,7 @@ public class TestSnapshots extends AbstractTestSnapshot {
   }
 
   @Override
-  protected void deleteSnapshot(String snapshotName) throws IOException {
+  protected void deleteSnapshot(String snapshotName) throws IOException, InterruptedException {
     try (Admin admin = getConnection().getAdmin()) {
       admin.deleteSnapshot(snapshotName);
     }
@@ -105,28 +136,6 @@ public class TestSnapshots extends AbstractTestSnapshot {
     }
   }
 
-  @Override
-  protected void deleteSnapshots(Pattern pattern) throws IOException {
-    try (Admin admin = getConnection().getAdmin()) {
-      admin.deleteSnapshots(pattern);
-    }
-  }
-
-  @Override
-  protected void deleteTableSnapshots(Pattern tableName, Pattern snapshotName) throws IOException {
-    try (Admin admin = getConnection().getAdmin()) {
-      admin.deleteTableSnapshots(tableName, snapshotName);
-    }
-  }
-
-  @Override
-  protected int listTableSnapshotsSize(String tableNameRegex, String snapshotNameRegex)
-      throws IOException {
-    try (Admin admin = getConnection().getAdmin()) {
-      return admin.listTableSnapshots(tableNameRegex, snapshotNameRegex).size();
-    }
-  }
-
   @Override
   protected int listSnapshotsSize(Pattern pattern) throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
@@ -134,14 +143,6 @@ public class TestSnapshots extends AbstractTestSnapshot {
     }
   }
 
-  @Override
-  protected int listTableSnapshotsSize(Pattern tableNamePattern, Pattern snapshotNamePattern)
-      throws IOException {
-    try (Admin admin = getConnection().getAdmin()) {
-      return admin.listTableSnapshots(tableNamePattern, snapshotNamePattern).size();
-    }
-  }
-
   @Override
   protected void deleteTable(TableName tableName) throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
@@ -149,10 +150,24 @@ public class TestSnapshots extends AbstractTestSnapshot {
     }
   }
 
-  @Override
-  protected int listTableSnapshotsSize(Pattern tableNamePattern) throws Exception {
+  protected Map<String, Long> createAndPopulateTable(TableName tableName) throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      return admin.listTableSnapshots(tableNamePattern, Pattern.compile(".*")).size();
+      HTableDescriptor descriptor = new HTableDescriptor(tableName);
+      descriptor.addFamily(new HColumnDescriptor(COLUMN_FAMILY));
+      admin.createTable(descriptor);
+
+      Map<String, Long> values = new HashMap<>();
+      try (Table table = getConnection().getTable(tableName)) {
+        List<Put> puts = new ArrayList<>();
+        for (long i = 0; i < 10; i++) {
+          final UUID rowKey = UUID.randomUUID();
+          byte[] row = Bytes.toBytes(rowKey.toString());
+          values.put(rowKey.toString(), i);
+          puts.add(new Put(row).addColumn(COLUMN_FAMILY, QUALIFIER, Bytes.toBytes(i)));
+        }
+        table.put(puts);
+      }
+      return values;
     }
   }
 }
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
index ce7100c10..e837ad093 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-hbase-2.x-integration-tests</artifactId>
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java
index 1cab6c56a..ea41eb8e2 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestSnapshots.java
@@ -15,104 +15,153 @@
  */ package com.google.cloud.bigtable.hbase;
 
 import static com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule.COLUMN_FAMILY;
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.MatcherAssert.assertThat;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.SnapshotDescription;
+import org.apache.hadoop.hbase.client.SnapshotType;
+import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
 
 public class TestSnapshots extends AbstractTestSnapshot {
 
-  @Override
-  protected void createTable(TableName tableName) throws IOException {
+  @Before
+  public void setUp() throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      HTableDescriptor descriptor = new HTableDescriptor(tableName);
-      descriptor.addFamily(new HColumnDescriptor(COLUMN_FAMILY));
-      admin.createTable(descriptor);
+      // Setup a prefix to avoid collisions between concurrent test runs
+      prefix = String.format("020%d", System.currentTimeMillis());
+
+      // clean up stale backups
+      String stalePrefix =
+          String.format("020%d", System.currentTimeMillis() - TimeUnit.HOURS.toMillis(2));
+
+      for (SnapshotDescription snapshotDescription : admin.listSnapshots()) {
+        int i = snapshotDescription.getName().lastIndexOf("/");
+        String backupId = snapshotDescription.getName().substring(i + 1);
+        if (backupId.endsWith(TEST_BACKUP_SUFFIX) && stalePrefix.compareTo(backupId) > 0) {
+          LOG.info("Deleting old snapshot: " + backupId);
+          admin.deleteSnapshot(backupId);
+        }
+      }
+      values = createAndPopulateTable(tableName);
     }
   }
 
-  @Override
-  protected void snapshot(String snapshotName, TableName tableName) throws IOException {
-    try (Admin admin = getConnection().getAdmin()) {
-      admin.snapshot(snapshotName, tableName);
+  @Test
+  public void testSnapshotDescription() throws IOException, InterruptedException {
+    String snapshotName = generateId("test-snapshot");
+    try {
+      SnapshotDescription snapshotDescription =
+          new SnapshotDescription(snapshotName, tableName, SnapshotType.FLUSH);
+      snapshot(snapshotDescription);
+      Assert.assertEquals(1, listSnapshotsSize(snapshotName));
+    } finally {
+      deleteSnapshot(snapshotName);
+      Assert.assertEquals(0, listSnapshotsSize(snapshotName));
     }
   }
 
-  @Override
-  protected int listSnapshotsSize(String regEx) throws IOException {
+  @Test
+  public void listSnapshots() throws IOException, InterruptedException {
+    String snapshot1 = generateId("snapshot-1");
+    snapshot(snapshot1, tableName);
+
+    String snapshot2 = generateId("snapshot-2");
+    snapshot(snapshot2, tableName);
+
     try (Admin admin = getConnection().getAdmin()) {
-      return admin.listSnapshots(regEx).size();
+      List<SnapshotDescription> snapshotDescriptions = admin.listSnapshots();
+      List<String> descStrings = new ArrayList<>();
+      for (SnapshotDescription snapshotDescription : snapshotDescriptions) {
+        descStrings.add(snapshotDescription.getName());
+      }
+      assertThat(descStrings, hasItems(snapshot1, snapshot2));
+    } finally {
+      deleteSnapshot(snapshot1);
+      deleteSnapshot(snapshot2);
     }
   }
 
   @Override
-  protected void deleteSnapshot(String snapshotName) throws IOException {
+  protected void snapshot(String snapshotName, TableName tableName)
+      throws IOException, InterruptedException {
     try (Admin admin = getConnection().getAdmin()) {
-      admin.deleteSnapshot(snapshotName);
+      admin.snapshot(snapshotName, tableName);
     }
   }
 
-  @Override
-  protected boolean tableExists(TableName tableName) throws IOException {
+  protected void snapshot(SnapshotDescription snapshotDescription)
+      throws IOException, InterruptedException {
     try (Admin admin = getConnection().getAdmin()) {
-      return admin.tableExists(tableName);
+      admin.snapshot(snapshotDescription);
     }
   }
 
   @Override
-  protected void disableTable(TableName tableName) throws IOException {
+  protected int listSnapshotsSize() throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      admin.disableTable(tableName);
+      return admin.listSnapshots().size();
     }
   }
 
   @Override
-  protected void cloneSnapshot(String snapshotName, TableName tableName)
-      throws IOException, TableExistsException, RestoreSnapshotException {
+  protected int listSnapshotsSize(String regEx) throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      admin.cloneSnapshot(snapshotName, tableName);
+      return admin.listSnapshots(regEx).size();
     }
   }
 
   @Override
-  protected void deleteSnapshots(Pattern pattern) throws IOException {
+  protected void deleteSnapshot(String snapshotName) throws IOException, InterruptedException {
     try (Admin admin = getConnection().getAdmin()) {
-      admin.deleteSnapshots(pattern);
+      admin.deleteSnapshot(snapshotName);
     }
   }
 
   @Override
-  protected void deleteTableSnapshots(Pattern tableName, Pattern snapshotName) throws IOException {
+  protected boolean tableExists(TableName tableName) throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      admin.deleteTableSnapshots(tableName, snapshotName);
+      return admin.tableExists(tableName);
     }
   }
 
   @Override
-  protected int listTableSnapshotsSize(String tableNameRegex, String snapshotNameRegex)
-      throws IOException {
+  protected void disableTable(TableName tableName) throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      return admin.listTableSnapshots(tableNameRegex, snapshotNameRegex).size();
+      admin.disableTable(tableName);
     }
   }
 
   @Override
-  protected int listSnapshotsSize(Pattern pattern) throws IOException {
+  protected void cloneSnapshot(String snapshotName, TableName tableName)
+      throws IOException, TableExistsException, RestoreSnapshotException {
     try (Admin admin = getConnection().getAdmin()) {
-      return admin.listSnapshots(pattern).size();
+      admin.cloneSnapshot(snapshotName, tableName);
     }
   }
 
   @Override
-  protected int listTableSnapshotsSize(Pattern tableNamePattern, Pattern snapshotNamePattern)
-      throws IOException {
+  protected int listSnapshotsSize(Pattern pattern) throws IOException {
     try (Admin admin = getConnection().getAdmin()) {
-      return admin.listTableSnapshots(tableNamePattern, snapshotNamePattern).size();
+      return admin.listSnapshots(pattern).size();
     }
   }
 
@@ -123,10 +172,24 @@ public class TestSnapshots extends AbstractTestSnapshot {
     }
   }
 
-  @Override
-  protected int listTableSnapshotsSize(Pattern tableNamePattern) throws Exception {
+  protected Map<String, Long> createAndPopulateTable(TableName tableName) throws IOException {
+    Map<String, Long> values = new HashMap<>();
     try (Admin admin = getConnection().getAdmin()) {
-      return admin.listTableSnapshots(tableNamePattern, Pattern.compile(".*")).size();
+      HTableDescriptor descriptor = new HTableDescriptor(tableName);
+      descriptor.addFamily(new HColumnDescriptor(COLUMN_FAMILY));
+      admin.createTable(descriptor);
+
+      try (Table table = getConnection().getTable(tableName)) {
+        List<Put> puts = new ArrayList<>();
+        for (long i = 0; i < 10; i++) {
+          final UUID rowKey = UUID.randomUUID();
+          byte[] row = Bytes.toBytes(rowKey.toString());
+          values.put(rowKey.toString(), i);
+          puts.add(new Put(row).addColumn(COLUMN_FAMILY, QUALIFIER, Bytes.toBytes(i)));
+        }
+        table.put(puts);
+      }
     }
+    return values;
   }
 }
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncSnapshots.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncSnapshots.java
index 12deabc12..8b321c389 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncSnapshots.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/async/TestAsyncSnapshots.java
@@ -16,6 +16,9 @@
 
 package com.google.cloud.bigtable.hbase.async;
 
+import static com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule.COLUMN_FAMILY;
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
@@ -23,6 +26,11 @@ import static org.junit.Assert.assertTrue;
 import com.google.cloud.bigtable.hbase.AbstractTestSnapshot;
 import com.google.cloud.bigtable.hbase.test_env.SharedTestEnvRule;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
@@ -30,9 +38,14 @@ import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.AsyncAdmin;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.SnapshotDescription;
+import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -40,6 +53,27 @@ import org.junit.runners.JUnit4;
 @RunWith(JUnit4.class)
 public class TestAsyncSnapshots extends AbstractTestSnapshot {
 
+  @Before
+  public void setUp() throws ExecutionException, InterruptedException, IOException {
+    // Setup a prefix to avoid collisions between concurrent test runs
+    prefix = String.format("020%d", System.currentTimeMillis());
+
+    // clean up stale backups
+    String stalePrefix =
+        String.format("020%d", System.currentTimeMillis() - TimeUnit.HOURS.toMillis(2));
+
+    for (SnapshotDescription snapshotDescription : getAsyncAdmin().listSnapshots().get()) {
+      int i = snapshotDescription.getName().lastIndexOf("/");
+      String backupId = snapshotDescription.getName().substring(i + 1);
+      if (backupId.endsWith(TEST_BACKUP_SUFFIX) && stalePrefix.compareTo(backupId) > 0) {
+        LOG.info("Deleting old snapshot: " + backupId);
+        getAsyncAdmin().deleteSnapshot(backupId);
+      }
+    }
+
+    values = createAndPopulateTable(tableName);
+  }
+
   private AsyncAdmin getAsyncAdmin() throws InterruptedException, ExecutionException {
     return AbstractAsyncTest.getAsyncConnection().getAdmin();
   }
@@ -68,61 +102,22 @@ public class TestAsyncSnapshots extends AbstractTestSnapshot {
   }
 
   @Test
-  public void testListTableSnapshotsWithNullAndEmptyString() throws IOException {
-    Exception actualError = null;
-    try {
-      listTableSnapshotsSize(null, "");
-    } catch (Exception ex) {
-      actualError = ex;
-    }
-    assertNotNull(actualError);
-    assertTrue(actualError instanceof NullPointerException);
-    actualError = null;
-
-    try {
-      listTableSnapshotsSize((Pattern) null, null);
-    } catch (Exception ex) {
-      actualError = ex;
-    }
-    assertNotNull(actualError);
-    assertTrue(actualError instanceof NullPointerException);
-
-    assertEquals(0, listTableSnapshotsSize("", ""));
-  }
-
-  @Test
-  public void testDeleteSnapshotWithEmptyString() throws Exception {
-    Exception actualError = null;
-    try {
-      // NPE is expected with AsyncAdmin.
-      deleteSnapshots(null);
-    } catch (Exception ex) {
-      actualError = ex;
-    }
-    assertNotNull(actualError);
-    assertTrue(actualError instanceof NullPointerException);
-    actualError = null;
-
-    try {
-      // NPE is expected with AsyncAdmin.
-      deleteTableSnapshots(null, null);
-    } catch (Exception ex) {
-      actualError = ex;
-    }
-    assertNotNull(actualError);
-
-    // No snapshot matches hence no exception should be thrown
-    deleteSnapshots(Pattern.compile(""));
+  public void listSnapshots() throws IOException, InterruptedException, ExecutionException {
+    String snapshot1 = generateId("snapshot-1");
+    snapshot(snapshot1, tableName);
 
-    deleteTableSnapshots(Pattern.compile(""), Pattern.compile(""));
-  }
-
-  @Override
-  protected void createTable(TableName tableName) throws IOException {
+    String snapshot2 = generateId("snapshot-2");
+    snapshot(snapshot2, tableName);
     try {
-      getAsyncAdmin().createTable(createDescriptor(tableName)).get();
-    } catch (InterruptedException | ExecutionException e) {
-      throw new IOException("Error while creating table: " + e.getCause());
+      List<SnapshotDescription> snapshotDescriptions = getAsyncAdmin().listSnapshots().get();
+      List<String> descStrings = new ArrayList<>();
+      for (SnapshotDescription snapshotDescription : snapshotDescriptions) {
+        descStrings.add(snapshotDescription.getName());
+      }
+      assertThat(descStrings, hasItems(snapshot1, snapshot2));
+    } finally {
+      deleteSnapshot(snapshot1);
+      deleteSnapshot(snapshot2);
     }
   }
 
@@ -142,6 +137,27 @@ public class TestAsyncSnapshots extends AbstractTestSnapshot {
     }
   }
 
+  protected Map<String, Long> createAndPopulateTable(TableName tableName) throws IOException {
+    try {
+      getAsyncAdmin().createTable(createDescriptor(tableName)).get();
+    } catch (InterruptedException | ExecutionException e) {
+      throw new IOException("Error while creating table: " + e.getCause());
+    }
+
+    Map<String, Long> values = new HashMap<>();
+    try (Table table = getConnection().getTable(tableName)) {
+      List<Put> puts = new ArrayList<>();
+      for (long i = 0; i < 10; i++) {
+        final UUID rowKey = UUID.randomUUID();
+        byte[] row = Bytes.toBytes(rowKey.toString());
+        values.put(rowKey.toString(), i);
+        puts.add(new Put(row).addColumn(COLUMN_FAMILY, QUALIFIER, Bytes.toBytes(i)));
+      }
+      table.put(puts);
+    }
+    return values;
+  }
+
   @Override
   protected void deleteSnapshot(String snapshotName) throws IOException {
     try {
@@ -179,24 +195,6 @@ public class TestAsyncSnapshots extends AbstractTestSnapshot {
     }
   }
 
-  @Override
-  protected void deleteSnapshots(Pattern pattern) throws IOException {
-    try {
-      getAsyncAdmin().deleteSnapshots(pattern).get();
-    } catch (InterruptedException | ExecutionException e) {
-      throw new IOException("Error while deleting snapshots: " + e.getCause());
-    }
-  }
-
-  @Override
-  protected void deleteTableSnapshots(Pattern tableName, Pattern snapshotRegEx) throws IOException {
-    try {
-      getAsyncAdmin().deleteTableSnapshots(tableName, snapshotRegEx).get();
-    } catch (InterruptedException | ExecutionException e) {
-      throw new IOException("Error while deleting snapshots: " + e.getCause());
-    }
-  }
-
   @Override
   protected int listSnapshotsSize(String regEx) throws IOException {
     try {
@@ -208,14 +206,11 @@ public class TestAsyncSnapshots extends AbstractTestSnapshot {
   }
 
   @Override
-  protected int listTableSnapshotsSize(String tableNameRegex, String snapshotNameRegex)
-      throws IOException {
+  protected int listSnapshotsSize() throws IOException {
     try {
-      Pattern tableNamePattern = Pattern.compile(tableNameRegex);
-      Pattern snapshotNamePattern = Pattern.compile(snapshotNameRegex);
-      return getAsyncAdmin().listTableSnapshots(tableNamePattern, snapshotNamePattern).get().size();
+      return getAsyncAdmin().listSnapshots().get().size();
     } catch (InterruptedException | ExecutionException e) {
-      throw new IOException("Error while listing table snapshot size: " + e.getCause());
+      throw new IOException("Error while listing snapshots size: " + e.getCause());
     }
   }
 
@@ -228,16 +223,6 @@ public class TestAsyncSnapshots extends AbstractTestSnapshot {
     }
   }
 
-  @Override
-  protected int listTableSnapshotsSize(Pattern tableNamePattern, Pattern snapshotNamePattern)
-      throws IOException {
-    try {
-      return getAsyncAdmin().listTableSnapshots(tableNamePattern, snapshotNamePattern).get().size();
-    } catch (InterruptedException | ExecutionException e) {
-      throw new IOException("Error while listing table snapshot size: " + e.getCause());
-    }
-  }
-
   @Override
   protected void deleteTable(TableName tableName) throws IOException {
     try {
@@ -246,13 +231,4 @@ public class TestAsyncSnapshots extends AbstractTestSnapshot {
       throw new IOException("Error while deleting table: " + e.getCause());
     }
   }
-
-  @Override
-  protected int listTableSnapshotsSize(Pattern tableNamePattern) throws Exception {
-    try {
-      return getAsyncAdmin().listTableSnapshots(tableNamePattern).get(60, TimeUnit.SECONDS).size();
-    } catch (InterruptedException | ExecutionException e) {
-      throw new IOException("Error while listing table snapshots: " + e.getCause());
-    }
-  }
 }
diff --git a/bigtable-test/bigtable-emulator-maven-plugin/pom.xml b/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
index 3ce4cdb5f..37e5260b4 100644
--- a/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
+++ b/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-test</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
 
diff --git a/bigtable-test/pom.xml b/bigtable-test/pom.xml
index d1fe2b3db..92a5ba00d 100644
--- a/bigtable-test/pom.xml
+++ b/bigtable-test/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-client-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.14.2-SNAPSHOT</version>
+    <version>1.15.1-SNAPSHOT</version>
   </parent>
 
   <artifactId>bigtable-test</artifactId>