Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1747 (2018-03-21 19:28:56)

Add ability to create a new presplit table for import
The new command is:

java -jar bigtable-beam-import-1.3.0-shaded.jar create-table \
    --project=[PROJECT_ID] \
    --bigtableInstanceId=[INSTANCE_ID] \
    --bigtableTableId=[TABLE_ID] \
    --sourcePattern='gs://[BUCKET_NAME]/[EXPORT_PATH]/part-*' \
    --families=[FAMILY1],[FAMILY2],...

Given a list of column families and the path of sequence files it will create a new empty table that is presplit appropriately for the import.  


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/pom.xml b/bigtable-client-core-parent/bigtable-client-core/pom.xml
index 2fd3661ec..446c09c55 100644
--- a/bigtable-client-core-parent/bigtable-client-core/pom.xml
+++ b/bigtable-client-core-parent/bigtable-client-core/pom.xml
@@ -107,6 +107,11 @@ limitations under the License.
             <artifactId>guava</artifactId>
             <version>${guava.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.threeten</groupId>
+            <artifactId>threetenbp</artifactId>
+            <version>${threetenbp.version}</version>
+        </dependency>
 
         <!-- Auth -->
         <dependency>
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCache.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCache.java
index f2b2cec75..92db299d6 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCache.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCache.java
@@ -33,6 +33,7 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import javax.annotation.concurrent.GuardedBy;
+import org.threeten.bp.Duration;
 
 /**
  * This class caches calls to {@link OAuth2Credentials#refreshAccessToken()}.  It asynchronously refreshes
@@ -181,9 +182,9 @@ public class OAuthCredentialsCache {
     return headerCache;
   }
 
-  HeaderToken getHeader(long timeoutMs) {
+  HeaderToken getHeader(Duration timeout) {
     try {
-      return getHeaderUnsafe(timeoutMs).getToken();
+      return getHeaderUnsafe(timeout).getToken();
     } catch (Exception e) {
       LOG.warn("Got an unexpected exception while trying to refresh google credentials.", e);
       Status status = Status.UNAUTHENTICATED
@@ -195,9 +196,9 @@ public class OAuthCredentialsCache {
 
   /**
    * Get the http credential header we need from a new oauth2 AccessToken.
-   * @param timeoutMs
    */
-  private HeaderCacheElement getHeaderUnsafe(long timeoutMs) {
+  @VisibleForTesting
+  HeaderCacheElement getHeaderUnsafe(Duration timeout) {
     // Optimize for the common case: do a volatile read to peek for a Good cache value
     HeaderCacheElement headerCacheUnsync = this.headerCache;
 
@@ -211,7 +212,7 @@ public class OAuthCredentialsCache {
       case Expired:
       case Exception:
         // defer the future resolution (asyncRefresh will spin up a thread that will try to acquire the lock)
-        return syncRefresh(timeoutMs);
+        return syncRefresh(timeout);
       default:
         String message = "Could not process state: " + headerCacheUnsync.getCacheState();
         LOG.warn(message);
@@ -226,9 +227,9 @@ public class OAuthCredentialsCache {
    * been refreshed.
    * This method should not be called while holding the refresh lock
    */
-  private HeaderCacheElement syncRefresh(long timeoutMs) {
+  private HeaderCacheElement syncRefresh(Duration timeout) {
     try (Closeable ss = Tracing.getTracer().spanBuilder("CredentialsRefresh").startScopedSpan()) {
-      return asyncRefresh().get(timeoutMs, TimeUnit.MILLISECONDS);
+      return asyncRefresh().get(timeout.toMillis(), TimeUnit.MILLISECONDS);
     } catch (InterruptedException e) {
       LOG.warn("Interrupted while trying to refresh google credentials.", e);
       Thread.currentThread().interrupt();
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptor.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptor.java
index 7b7817983..b2b7d66b0 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptor.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptor.java
@@ -33,6 +33,7 @@ import io.grpc.Status;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
+import org.threeten.bp.Duration;
 
 /**
  * Client interceptor that authenticates all calls by binding header data provided by a credential.
@@ -54,7 +55,7 @@ import java.util.concurrent.TimeUnit;
 public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor {
   private static final Logger LOG = new Logger(RefreshingOAuth2CredentialsInterceptor.class);
 
-  public static int TIMEOUT_SECONDS = 15;
+  public static Duration TIMEOUT = Duration.ofSeconds(15);
 
   @VisibleForTesting
   static final Metadata.Key<String> AUTHORIZATION_HEADER_KEY = Metadata.Key.of(
@@ -120,13 +121,13 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
 
       @Override
       public void start(Listener<RespT> responseListener, Metadata headers) {
-        long timeoutMs = -1;
+        final Duration timeout;
         if (callOptions.getDeadline() != null) {
-          timeoutMs = callOptions.getDeadline().timeRemaining(TimeUnit.MILLISECONDS);
+          timeout = Duration.ofMillis(callOptions.getDeadline().timeRemaining(TimeUnit.MILLISECONDS));
         } else {
-          timeoutMs = TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS);
+          timeout = TIMEOUT;
         }
-        HeaderToken token = store.getHeader(timeoutMs);
+        HeaderToken token = store.getHeader(timeout);
 
         if (!token.getStatus().isOk()) {
           unauthorized = true;
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
index c11f80a42..be7f8a324 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
+++ b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
@@ -103,6 +103,15 @@ limitations under the License.
 
   <build>
     <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <source>1.8</source>
+          <target>1.8</target>
+        </configuration>
+      </plugin>
+
       <plugin>
         <artifactId>maven-jar-plugin</artifactId>
         <configuration>
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/CreateTableHelper.java b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/CreateTableHelper.java
new file mode 100644
index 000000000..a5feffee1
--- /dev/null
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/CreateTableHelper.java
@@ -0,0 +1,113 @@
+/*
+ * Copyright 2018 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
+ *
+ *     https://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.beam.sequencefiles;
+
+import com.google.cloud.bigtable.beam.sequencefiles.ImportJob.ImportOptions;
+import com.google.cloud.bigtable.hbase.BigtableConfiguration;
+import com.google.common.base.Throwables;
+import java.util.List;
+import java.util.Objects;
+import org.apache.beam.sdk.io.BoundedSource.BoundedReader;
+import org.apache.beam.sdk.io.FileBasedSource;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+
+/**
+ * Command to create a new table.
+ *
+ * <p>The table will be presplit based on row keys collected from a set of sequence files. This is
+ * intended to be a preparation step before running an {@link ImportJob}.
+ */
+class CreateTableHelper {
+  private static final Log LOG = LogFactory.getLog(CreateTableHelper.class);
+
+  interface CreateTableOpts extends ImportOptions {
+    @Description("The families to add to the new table")
+    List<String> getFamilies();
+    void setFamilies(List<String> families);
+  }
+
+  public static void main(String[] args) throws Exception {
+    PipelineOptionsFactory.register(CreateTableOpts.class);
+
+    CreateTableOpts opts = PipelineOptionsFactory
+        .fromArgs(args).withValidation()
+        .as(CreateTableOpts.class);
+
+    FileSystems.setDefaultPipelineOptions(opts);
+
+    createTable(opts);
+
+  }
+
+  static void createTable(CreateTableOpts opts) throws Exception {
+    LOG.info("Extracting splits from the source files");
+
+    // Create the same source as the import job would use
+    SequenceFileSource<ImmutableBytesWritable, Result> source = ImportJob.createSource(opts);
+
+    // Extract the splits from the sequence files
+    List<? extends FileBasedSource<KV<ImmutableBytesWritable, Result>>> splitSources = source
+        .split(ImportJob.BUNDLE_SIZE, opts);
+
+    // Read the start key of each split
+    byte[][] splits = splitSources.stream()
+        .parallel()
+        .map(splitSource -> {
+          try (BoundedReader<KV<ImmutableBytesWritable, Result>> reader = splitSource
+              .createReader(opts)) {
+            if (reader.start()) {
+              return reader.getCurrent().getKey();
+            }
+          } catch (Exception e) {
+            Throwables.propagate(e);
+          }
+          return null;
+        })
+        .filter(Objects::nonNull)
+        .sorted()
+        .map(ImmutableBytesWritable::copyBytes)
+        .toArray(byte[][]::new);
+
+    LOG.info(String.format("Creating a new table with %d splits and the families: %s",
+        splits.length, opts.getFamilies()));
+
+    try (Connection connection = BigtableConfiguration
+        .connect(opts.getBigtableProject(), opts.getBigtableInstanceId())) {
+
+      TableName tableName = TableName.valueOf(opts.getBigtableTableId());
+      HTableDescriptor descriptor = new HTableDescriptor(tableName);
+
+      for (String family : opts.getFamilies()) {
+        descriptor.addFamily(
+            new HColumnDescriptor(family)
+        );
+      }
+
+      connection.getAdmin().createTable(descriptor, splits);
+    }
+  }
+}
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ImportJob.java b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ImportJob.java
index 41b93ed3d..2f5307617 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ImportJob.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ImportJob.java
@@ -22,14 +22,19 @@ import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.PipelineResult.State;
 import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.io.Read;
 import org.apache.beam.sdk.options.Default;
 import org.apache.beam.sdk.options.Description;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.io.serializer.WritableSerialization;
@@ -37,6 +42,8 @@ import org.apache.hadoop.io.serializer.WritableSerialization;
 public class ImportJob {
   private static final Log LOG = LogFactory.getLog(ImportJob.class);
 
+  static final long BUNDLE_SIZE = 100 * 1024 * 1024;
+
   public interface ImportOptions extends GcpOptions {
     //TODO: switch to ValueProviders
 
@@ -97,13 +104,24 @@ public class ImportJob {
   static Pipeline buildPipeline(ImportOptions opts) {
     Pipeline pipeline = Pipeline.create(Utils.tweakOptions(opts));
 
-    SequenceFileSource<ImmutableBytesWritable, Result> source = new SequenceFileSource<>(
+    pipeline
+        .apply("Read Sequence File", Read.from(new ShuffledSource<>(createSource(opts))))
+        .apply("Create Mutations", ParDo.of(new HBaseResultToMutationFn()))
+        .apply("Write to Bigtable", createSink(opts));
+
+    return pipeline;
+  }
+
+  static SequenceFileSource<ImmutableBytesWritable, Result> createSource(ImportOptions opts) {
+    return new SequenceFileSource<>(
         opts.getSourcePattern(),
         ImmutableBytesWritable.class, WritableSerialization.class,
         Result.class, ResultSerialization.class,
-        100 * 1024 * 1024
+        BUNDLE_SIZE
     );
+  }
 
+  static PTransform<PCollection<Mutation>, PDone> createSink(ImportOptions opts) {
     CloudBigtableTableConfiguration.Builder configBuilder = new CloudBigtableTableConfiguration.Builder()
         .withProjectId(opts.getBigtableProject())
         .withInstanceId(opts.getBigtableInstanceId())
@@ -113,11 +131,6 @@ public class ImportJob {
       configBuilder.withAppProfileId(opts.getBigtableAppProfileId());
     }
 
-    pipeline
-        .apply("Read Sequence File", Read.from(new ShuffledSource<>(source)))
-        .apply("Create Mutations", ParDo.of(new HBaseResultToMutationFn()))
-        .apply("Write to Bigtable", CloudBigtableIO.writeToTable(configBuilder.build()));
-
-    return pipeline;
+    return CloudBigtableIO.writeToTable(configBuilder.build());
   }
 }
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/Main.java b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/Main.java
index 6a12dd755..d25f44a69 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/Main.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/Main.java
@@ -20,7 +20,7 @@ import java.net.URISyntaxException;
 import java.util.Arrays;
 
 /**
- * Entry point for import/export job submission.
+ * Entry point for create-table/import/export job submission.
  *
  * @author igorbernstein2
  */
@@ -40,6 +40,9 @@ public class Main {
       case "import":
         ImportJob.main(subArgs);
         break;
+      case "create-table":
+        CreateTableHelper.main(subArgs);
+        break;
       default:
         usage();
         System.exit(1);
@@ -58,7 +61,7 @@ public class Main {
 
     System.out.printf(
         "java -jar %s <action> <action_params>\n"
-            + "Where <action> can be 'export' or 'import'. To get further help, run: \n"
+            + "Where <action> can be 'export', 'import' or 'create-table'. To get further help, run: \n"
             + "java -jar %s <action> --help\n",
         jarName,
         jarName
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 a42667d9f..23c577e9c 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
@@ -146,6 +146,8 @@ limitations under the License.
                   <exclude>junit:junit</exclude>
                   <exclude>org.hamcrest:hamcrest-core</exclude>
                   <exclude>javax.inject:javax.inject</exclude>
+                  <exclude>org.apache.htrace:htrace-core4</exclude>
+                  <exclude>org.apache.yetus:audience-annotations</exclude>
                 </excludes>
               </artifactSet>
               <relocations>
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 5c5d74996..33fe3e55c 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
@@ -29,27 +29,21 @@ import java.util.regex.Pattern;
 
 import com.google.bigtable.admin.v2.*;
 import com.google.cloud.bigtable.hbase2_x.adapters.admin.TableAdapter2x;
-import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.apache.hadoop.hbase.CacheEvictionStats;
 import org.apache.hadoop.hbase.ClusterMetrics;
 import org.apache.hadoop.hbase.ClusterMetrics.Option;
-import org.apache.hadoop.hbase.ClusterStatus;
 import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.NamespaceDescriptor;
-import org.apache.hadoop.hbase.RegionLoad;
 import org.apache.hadoop.hbase.RegionMetrics;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.client.AbstractBigtableAdmin;
 import org.apache.hadoop.hbase.client.AbstractBigtableConnection;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
 import org.apache.hadoop.hbase.client.CompactType;
 import org.apache.hadoop.hbase.client.CompactionState;
-import org.apache.hadoop.hbase.client.MasterSwitchType;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.SnapshotDescription;
 import org.apache.hadoop.hbase.client.SnapshotType;
@@ -216,14 +210,8 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   /** {@inheritDoc} */
   @Override
-  public TableDescriptor getDescriptor(TableName tableName)
-      throws TableNotFoundException, IOException {
-    HTableDescriptor[] descriptors =
-        getTableDescriptorsByTableName(new ArrayList<TableName>(Arrays.asList(tableName)));
-    if (descriptors != null && descriptors.length > 0) {
-      return descriptors[0];
-    }
-    return null;
+  public TableDescriptor getDescriptor(TableName tableName) throws IOException {
+    return getTableDescriptor(tableName);
   }
 
   /** {@inheritDoc} */
@@ -257,7 +245,7 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
     Modification.Builder modification = Modification.newBuilder().setId(columnName)
         .setCreate(new ColumnDescriptorAdapter()
         .adapt((HColumnDescriptor) columnFamily).build());
-    return modifyColumnAsync(tableName, columnName, "add", modification);
+    return modifyColumnAsync(tableName, modification);
   }
 
   @Override
@@ -271,12 +259,11 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
     final String columnNameStr = Bytes.toString(columnName);
     Modification.Builder modification = Modification.newBuilder().setId(columnNameStr)
         .setDrop(true);
-    return modifyColumnAsync(tableName, columnNameStr, "delete", modification);
+    return modifyColumnAsync(tableName, modification);
   }
 
 
-  protected CompletableFuture<Void> modifyColumnAsync(TableName tableName, String columnName,
-      String modificationType, Modification.Builder modification) {
+  protected CompletableFuture<Void> modifyColumnAsync(TableName tableName, Modification.Builder modification) {
     ModifyColumnFamiliesRequest.Builder modifyColumnBuilder = ModifyColumnFamiliesRequest
         .newBuilder().addModifications(modification).setName(toBigtableName(tableName));
     return FutureUtils.toCompletableFuture(
@@ -351,14 +338,13 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
     return response;
   }
 
-
   @Override
   public Future<Void> modifyColumnFamilyAsync(TableName tableName, 
       ColumnFamilyDescriptor columnFamily) throws IOException {
     String columnName = columnFamily.getNameAsString();
     Modification.Builder modification = Modification.newBuilder().setId(columnName)
         .setUpdate(new ColumnDescriptorAdapter().adapt((HColumnDescriptor) columnFamily).build());
-    return modifyColumnAsync(tableName, columnName, "update", modification);
+    return modifyColumnAsync(tableName, modification);
   }
 
   @Override
@@ -390,6 +376,7 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
   }
 
   /* ******* Unsupported methods *********** */
+
   @Override
   public boolean abortProcedure(long arg0, boolean arg1) throws IOException {
     throw new UnsupportedOperationException("abortProcedure");
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 1a1d695f6..b2186f0c3 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
@@ -274,6 +274,96 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     });
   }
 
+  // ****** TO BE IMPLEMENTED [start] ******
+
+  @Override
+  public CompletableFuture<Void> deleteColumnFamily(TableName arg0, byte[] arg1) {
+    throw new UnsupportedOperationException("deleteColumnFamily"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteSnapshot(String arg0) {
+    throw new UnsupportedOperationException("deleteSnapshot"); // TODO
+
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteTableSnapshots(Pattern arg0, Pattern arg1) {
+    throw new UnsupportedOperationException("deleteTableSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> modifyColumnFamily(TableName arg0, ColumnFamilyDescriptor arg1) {
+    throw new UnsupportedOperationException("modifyColumnFamily"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> restoreSnapshot(String arg0) {
+    throw new UnsupportedOperationException("restoreSnapshot"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> restoreSnapshot(String arg0, boolean arg1) {
+    throw new UnsupportedOperationException("restoreSnapshot"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> truncateTable(TableName arg0, boolean arg1) {
+    throw new UnsupportedOperationException("truncateTable"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteSnapshots() {
+    throw new UnsupportedOperationException("deleteSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteSnapshots(Pattern arg0) {
+    throw new UnsupportedOperationException("deleteSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> deleteTableSnapshots(Pattern arg0) {
+    throw new UnsupportedOperationException("deleteTableSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> cloneSnapshot(String arg0, TableName arg1) {
+    throw new UnsupportedOperationException("cloneSnapshot"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<SnapshotDescription>> listSnapshots() {
+    throw new UnsupportedOperationException("listSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<SnapshotDescription>> listSnapshots(Pattern arg0) {
+    throw new UnsupportedOperationException("listSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern arg0) {
+    throw new UnsupportedOperationException("listTableSnapshots"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Void> modifyTable(TableDescriptor arg0) {
+    throw new UnsupportedOperationException("modifyTable"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<Boolean> isTableAvailable(TableName arg0) {
+    throw new UnsupportedOperationException("isTableAvailable"); // TODO
+  }
+
+  @Override
+  public CompletableFuture<List<RegionInfo>> getRegions(TableName arg0) {
+    throw new UnsupportedOperationException("getRegions"); // TODO
+  }
+
+  // ****** TO BE IMPLEMENTED [end] ******
+
   /** {@inheritDoc} */
   @Override
   public CompletableFuture<Boolean> abortProcedure(long arg0, boolean arg1) {
@@ -313,11 +403,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("clearDeadServers"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Void> cloneSnapshot(String arg0, TableName arg1) {
-    throw new UnsupportedOperationException("cloneSnapshot"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> compactRegionServer(ServerName arg0) {
     throw new UnsupportedOperationException("compactRegionServer"); // TODO
@@ -333,27 +418,12 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("decommissionRegionServers"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Void> deleteColumnFamily(TableName arg0, byte[] arg1) {
-    throw new UnsupportedOperationException("deleteColumnFamily"); // TODO
-  }
 
   @Override
   public CompletableFuture<Void> deleteNamespace(String arg0) {
     throw new UnsupportedOperationException("deleteNamespace"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Void> deleteSnapshot(String arg0) {
-    throw new UnsupportedOperationException("deleteSnapshot"); // TODO
-
-  }
-
-  @Override
-  public CompletableFuture<Void> deleteTableSnapshots(Pattern arg0, Pattern arg1) {
-    throw new UnsupportedOperationException("deleteTableSnapshots"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> disableReplicationPeer(String arg0) {
     throw new UnsupportedOperationException("disableReplicationPeer"); // TODO
@@ -490,11 +560,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("mergeRegions"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Void> modifyColumnFamily(TableName arg0, ColumnFamilyDescriptor arg1) {
-    throw new UnsupportedOperationException("modifyColumnFamily"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> modifyNamespace(NamespaceDescriptor arg0) {
     throw new UnsupportedOperationException("modifyNamespace"); // TODO
@@ -520,16 +585,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("removeReplicationPeer"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Void> restoreSnapshot(String arg0) {
-    throw new UnsupportedOperationException("restoreSnapshot"); // TODO
-  }
-
-  @Override
-  public CompletableFuture<Void> restoreSnapshot(String arg0, boolean arg1) {
-    throw new UnsupportedOperationException("restoreSnapshot"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> rollWALWriter(ServerName arg0) {
     throw new UnsupportedOperationException("rollWALWriter"); // TODO
@@ -580,11 +635,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("stopRegionServer"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Void> truncateTable(TableName arg0, boolean arg1) {
-    throw new UnsupportedOperationException("truncateTable"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> unassign(byte[] arg0, boolean arg1) {
     throw new UnsupportedOperationException("unassign"); // TODO
@@ -677,36 +727,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("coprocessorService"); // TODO
   }
 
-  /*
-   * This method should be implemented.
-   * (non-Javadoc)
-   * @see org.apache.hadoop.hbase.client.AsyncAdmin#deleteSnapshots()
-   */
-  @Override
-  public CompletableFuture<Void> deleteSnapshots() {
-    throw new UnsupportedOperationException("deleteSnapshots"); // TODO
-  }
-
-  /*
-   * This method should be implemented.
-   * (non-Javadoc)
-   * @see org.apache.hadoop.hbase.client.AsyncAdmin#deleteSnapshots(java.util.regex.Pattern)
-   */
-  @Override
-  public CompletableFuture<Void> deleteSnapshots(Pattern arg0) {
-    throw new UnsupportedOperationException("deleteSnapshots"); // TODO
-  }
-
-  /*
-   * This method should be implemented.
-   * (non-Javadoc)
-   * @see org.apache.hadoop.hbase.client.AsyncAdmin#deleteTableSnapshots(java.util.regex.Pattern)
-   */
-  @Override
-  public CompletableFuture<Void> deleteTableSnapshots(Pattern arg0) {
-    throw new UnsupportedOperationException("deleteTableSnapshots"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> disableTableReplication(TableName arg0) {
     throw new UnsupportedOperationException("disableTableReplication"); // TODO
@@ -753,11 +773,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("getRegions"); // TODO
   }
 
-  @Override
-  public CompletableFuture<List<RegionInfo>> getRegions(TableName arg0) {
-    throw new UnsupportedOperationException("getRegions"); // TODO
-  }
-
   @Override
   public CompletableFuture<Boolean> isBalancerEnabled() {
     throw new UnsupportedOperationException("isBalancerEnabled"); // TODO
@@ -788,11 +803,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("isSplitEnabled"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Boolean> isTableAvailable(TableName arg0) {
-    throw new UnsupportedOperationException("isTableAvailable"); // TODO
-  }
-
   @Override
   public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers() {
     throw new UnsupportedOperationException("listReplicationPeers"); // TODO
@@ -803,26 +813,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("listReplicationPeers"); // TODO
   }
 
-  /*
-   * This method should be implemented.
-   * (non-Javadoc)
-   * @see org.apache.hadoop.hbase.client.AsyncAdmin#listSnapshots()
-   */
-  @Override
-  public CompletableFuture<List<SnapshotDescription>> listSnapshots() {
-    throw new UnsupportedOperationException("listSnapshots"); // TODO
-  }
-
-  /*
-   * This method should be implemented.
-   * (non-Javadoc)
-   * @see org.apache.hadoop.hbase.client.AsyncAdmin#listSnapshots(java.util.regex.Pattern)
-   */
-  @Override
-  public CompletableFuture<List<SnapshotDescription>> listSnapshots(Pattern arg0) {
-    throw new UnsupportedOperationException("listSnapshots"); // TODO
-  }
-
   @Override
   public CompletableFuture<List<TableDescriptor>> listTableDescriptorsByNamespace(String arg0) {
     throw new UnsupportedOperationException("listTableDescriptorsByNamespace"); // TODO
@@ -833,11 +823,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("listTableNamesByNamespace"); // TODO
   }
 
-  @Override
-  public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern arg0) {
-    throw new UnsupportedOperationException("listTableSnapshots"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> majorCompact(TableName arg0, CompactType arg1) {
     throw new UnsupportedOperationException("majorCompact"); // TODO
@@ -863,11 +848,6 @@ public class BigtableAsyncAdmin implements AsyncAdmin {
     throw new UnsupportedOperationException("mergeSwitch"); // TODO
   }
 
-  @Override
-  public CompletableFuture<Void> modifyTable(TableDescriptor arg0) {
-    throw new UnsupportedOperationException("modifyTable"); // TODO
-  }
-
   @Override
   public CompletableFuture<Void> move(byte[] arg0) {
     throw new UnsupportedOperationException("move"); // TODO
diff --git a/bigtable-hbase-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java b/bigtable-hbase-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
index c70f06728..461cd47f7 100644
--- a/bigtable-hbase-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
+++ b/bigtable-hbase-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableAdmin.java
@@ -281,8 +281,7 @@ public abstract class AbstractBigtableAdmin implements Admin {
 
   /** {@inheritDoc} */
   @Override
-  public HTableDescriptor getTableDescriptor(TableName tableName)
-      throws TableNotFoundException, IOException {
+  public HTableDescriptor getTableDescriptor(TableName tableName) throws IOException {
     if (tableName == null) {
       return null;
     }
diff --git a/pom.xml b/pom.xml
index 0ffc32f59..842aaad8e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -65,6 +65,7 @@ limitations under the License.
         <guava.version>19.0</guava.version>
         <google.auth.library.version>0.9.0</google.auth.library.version>
         <dropwizard.metrics.version>3.1.2</dropwizard.metrics.version>
+        <threetenbp.version>1.3.3</threetenbp.version>
 
 
         <beam.version>2.3.0</beam.version>


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCacheTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCacheTest.java
index 510522493..1dc5c3ee8 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCacheTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/OAuthCredentialsCacheTest.java
@@ -34,6 +34,7 @@ import org.mockito.stubbing.Answer;
 import java.io.IOException;
 import java.util.Date;
 import java.util.concurrent.*;
+import org.threeten.bp.Duration;
 
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.mockito.Matchers.any;
@@ -45,7 +46,7 @@ import static org.mockito.Mockito.*;
  */
 @RunWith(JUnit4.class)
 public class OAuthCredentialsCacheTest {
-  private static int TIMEOUT_SECONDS = 5;
+  private static Duration TIMEOUT = Duration.ofSeconds(5);
 
   private static ExecutorService executorService;
 
@@ -102,7 +103,7 @@ public class OAuthCredentialsCacheTest {
     ExecutorService mockExecutor = Mockito.mock(ExecutorService.class);
     when(mockExecutor.submit(any(Callable.class))).thenThrow(new RuntimeException(""));
     underTest = new OAuthCredentialsCache(mockExecutor, mockCredentials);
-    Assert.assertFalse(underTest.getHeader(TIMEOUT_SECONDS).getStatus().isOk());
+    Assert.assertFalse(underTest.getHeader(TIMEOUT).getStatus().isOk());
   }
 
   @Test
@@ -147,11 +148,11 @@ public class OAuthCredentialsCacheTest {
         .thenReturn(accessToken);
 
     // First call
-    OAuthCredentialsCache.HeaderToken token1 = underTest.getHeader(TIMEOUT_SECONDS);
+    OAuthCredentialsCache.HeaderToken token1 = underTest.getHeader(TIMEOUT);
     Assert.assertFalse(token1.getStatus().isOk());
 
     // Now the second token should be available
-    OAuthCredentialsCache.HeaderToken token2 = underTest.getHeader(TIMEOUT_SECONDS);
+    OAuthCredentialsCache.HeaderToken token2 = underTest.getHeader(TIMEOUT);
     Assert.assertTrue(token2.getStatus().isOk());
     Assert.assertThat(token2.getHeader(), containsString("hi"));
     // Make sure that the token was only requested twice: once for the first failure & second time for background recovery
@@ -175,7 +176,7 @@ public class OAuthCredentialsCacheTest {
         .thenReturn(goodToken);
 
     // First call - setup
-    OAuthCredentialsCache.HeaderToken firstResult = underTest.getHeader(TIMEOUT_SECONDS);
+    OAuthCredentialsCache.HeaderToken firstResult = underTest.getHeader(TIMEOUT);
     Assert.assertEquals(CacheState.Good, underTest.getHeaderCache().getCacheState());
     Assert.assertThat(firstResult.getHeader(), containsString("stale"));
 
@@ -183,15 +184,15 @@ public class OAuthCredentialsCacheTest {
     setTimeInMillieconds(10);
 
     // Second call - return stale token, but schedule refresh
-    OAuthCredentialsCache.HeaderToken secondResult = underTest.getHeader(TIMEOUT_SECONDS);
-    Assert.assertEquals(CacheState.Stale, underTest.getHeaderCache().getCacheState());
-    Assert.assertThat(secondResult.getHeader(), containsString("stale"));
+    HeaderCacheElement secondResult = underTest.getHeaderUnsafe(TIMEOUT);
+    Assert.assertEquals(CacheState.Stale, secondResult.getCacheState());
+    Assert.assertThat(secondResult.getToken().getHeader(), containsString("stale"));
 
     // forward to expired
     setTimeInMillieconds(expires);
 
     // Third call - now returns good token
-    OAuthCredentialsCache.HeaderToken thirdResult = underTest.getHeader(TIMEOUT_SECONDS);
+    OAuthCredentialsCache.HeaderToken thirdResult = underTest.getHeader(TIMEOUT);
     Assert.assertEquals(CacheState.Good, underTest.getHeaderCache().getCacheState());
     Assert.assertThat(thirdResult.getHeader(), containsString("good"));
 
@@ -223,13 +224,13 @@ public class OAuthCredentialsCacheTest {
     underTest.revokeUnauthToken(underTest.getHeaderCache().getToken());
     Assert.assertEquals(CacheState.Expired, underTest.getHeaderCache().getCacheState());
 
-    underTest.getHeader(TIMEOUT_SECONDS);
+    underTest.getHeader(TIMEOUT);
     Assert.assertEquals(CacheState.Good, underTest.getHeaderCache().getCacheState());
 
     underTest.revokeUnauthToken(underTest.getHeaderCache().getToken());
     // Ensure that the revoke interceptor does not run so quickly after the previous revoke.
     // The cache state should stay good.
-    Assert.assertTrue(underTest.getHeader(5).getStatus().isOk());
+    Assert.assertTrue(underTest.getHeader(TIMEOUT).getStatus().isOk());
   }
 
   private void testHanging()
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptorTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptorTest.java
index 29cf25ab1..3b35aff7f 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptorTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/RefreshingOAuth2CredentialsInterceptorTest.java
@@ -40,6 +40,7 @@ import org.mockito.ArgumentCaptor;
 import org.mockito.Matchers;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.threeten.bp.Duration;
 
 
 /**
@@ -104,7 +105,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     public void testDefaultDeadline() throws IOException {
         initializeOk();
         sendRequest(CallOptions.DEFAULT);
-        assertEquals(RefreshingOAuth2CredentialsInterceptor.TIMEOUT_SECONDS * 1000, getDeadlineMs());
+        assertEquals(RefreshingOAuth2CredentialsInterceptor.TIMEOUT.toMillis(), getDeadlineMs());
     }
 
     @Test
@@ -128,9 +129,9 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     }
 
     private long getDeadlineMs() {
-        ArgumentCaptor<Long> timeoutCaptor = ArgumentCaptor.forClass(Long.class);
+        ArgumentCaptor<Duration> timeoutCaptor = ArgumentCaptor.forClass(Duration.class);
         verify(mockCache, times(1)).getHeader(timeoutCaptor.capture());
-        return timeoutCaptor.getValue();
+        return timeoutCaptor.getValue().toMillis();
     }
 
     @Test
@@ -141,7 +142,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
         Status grpcStatus = Status.UNAUTHENTICATED;
 
         // Something bad happened, and authentication could not be established
-        when(mockCache.getHeader(anyInt()))
+        when(mockCache.getHeader(any(Duration.class)))
                 .thenReturn(new OAuthCredentialsCache.HeaderToken(grpcStatus, null));
 
         ClientCall<ReadRowsRequest, ReadRowsResponse> call = underTest.interceptCall(
@@ -218,7 +219,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     }
 
     private void initializeOk() {
-        when(mockCache.getHeader(anyInt()))
+        when(mockCache.getHeader(any(Duration.class)))
                 .thenReturn(new OAuthCredentialsCache.HeaderToken(Status.OK, HEADER));
     }
 
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/EndToEndIT.java b/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/EndToEndIT.java
index f359230bc..bd1c78da4 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/EndToEndIT.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/EndToEndIT.java
@@ -17,10 +17,12 @@ package com.google.cloud.bigtable.beam.sequencefiles;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.google.cloud.bigtable.beam.sequencefiles.CreateTableHelper.CreateTableOpts;
 import com.google.cloud.bigtable.beam.sequencefiles.ExportJob.ExportOptions;
 import com.google.cloud.bigtable.beam.sequencefiles.testing.BigtableTableUtils;
 import com.google.cloud.bigtable.hbase.BigtableConfiguration;
 import com.google.cloud.bigtable.hbase.BigtableOptionsFactory;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Sets;
 import java.io.File;
 import java.io.IOException;
@@ -119,7 +121,7 @@ public class EndToEndIT {
   }
 
   @Test
-  public void testExportImport() throws IOException {
+  public void testExportImport() throws Exception {
     // Create a table, populate it & export it
     final List<Put> testData = Arrays.asList(
         new Put(Bytes.toBytes("row_key_1"))
@@ -167,7 +169,7 @@ public class EndToEndIT {
     final String destTableId = tableId + "-verify";
 
     try (BigtableTableUtils destTable = new BigtableTableUtils(connection, destTableId, CF)) {
-      destTable.createEmptyTable();
+      destTable.deleteTable();
 
       DataflowPipelineOptions pipelineOpts = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
       pipelineOpts.setRunner(DataflowRunner.class);
@@ -180,6 +182,10 @@ public class EndToEndIT {
       importOpts.setBigtableTableId(destTableId);
       importOpts.setSourcePattern(StaticValueProvider.of(workDir + "/part-*"));
 
+      CreateTableHelper.CreateTableOpts createOpts = pipelineOpts.as(CreateTableHelper.CreateTableOpts.class);
+      createOpts.setFamilies(ImmutableList.of(CF));
+      CreateTableHelper.createTable(createOpts);
+
       State state = ImportJob.buildPipeline(importOpts).run().waitUntilFinish();
       Assert.assertEquals(State.DONE, state);
 
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/testing/BigtableTableUtils.java b/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/testing/BigtableTableUtils.java
index 52fbd0dbb..a4012e102 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/testing/BigtableTableUtils.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/testing/BigtableTableUtils.java
@@ -67,6 +67,14 @@ public class BigtableTableUtils implements AutoCloseable {
     admin.createTable(tableDescriptor);
   }
 
+  /** Delete the table if it exists */
+  public void deleteTable() throws IOException {
+    if (admin.tableExists(tableName)) {
+      admin.disableTable(tableName);
+      admin.deleteTable(tableName);
+    }
+  }
+
   /**
    * Returns true if table already exists.
    */