Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1720 (2018-03-09 21:10:01)

Upgrade to Beam 2.3
* Fix destination resource id generation
* Upgrade to 2.3
* Update SequenceFileIO to use a constant dynamic destination



Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceClient.java
index 2234f7750..bbceb0d1e 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableInstanceClient.java
@@ -15,21 +15,36 @@
  */
 package com.google.cloud.bigtable.grpc;
 
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import com.google.bigtable.admin.v2.AppProfile;
 import com.google.bigtable.admin.v2.Cluster;
+import com.google.bigtable.admin.v2.CreateAppProfileRequest;
 import com.google.bigtable.admin.v2.CreateInstanceRequest;
+import com.google.bigtable.admin.v2.DeleteAppProfileRequest;
+import com.google.bigtable.admin.v2.DeleteClusterRequest;
 import com.google.bigtable.admin.v2.DeleteInstanceRequest;
+import com.google.bigtable.admin.v2.GetAppProfileRequest;
 import com.google.bigtable.admin.v2.GetClusterRequest;
 import com.google.bigtable.admin.v2.Instance;
+import com.google.bigtable.admin.v2.ListAppProfilesRequest;
+import com.google.bigtable.admin.v2.ListAppProfilesResponse;
 import com.google.bigtable.admin.v2.ListClustersRequest;
 import com.google.bigtable.admin.v2.ListClustersResponse;
 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.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;
 
 /**
  * BigtableInstanceClient manages instances and clusters.
@@ -128,4 +143,121 @@ public interface BigtableInstanceClient {
    * @return a {@link com.google.longrunning.Operation} object.
    */
   Operation updateCluster(Cluster cluster);
+  
+  /**
+   * Deletes a cluster from an instance.
+   * 
+   * @param request a {@link com.google.bigtable.admin.v2.DeleteClusterRequest} object.
+   * @return a {@link com.google.protobuf.Empty} object.
+   */
+  Empty deleteCluster(DeleteClusterRequest request);
+
+  /**
+   * Partially updates an instance within a project.
+   * 
+   * @param request a {@link com.google.bigtable.admin.v2.PartialUpdateInstanceRequest} object.
+   * @return a {@link com.google.longrunning.Operation} object.
+   */
+  Operation partialUpdateInstance(PartialUpdateInstanceRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable replication. This feature
+   * is not currently available to most Cloud Bigtable customers. This feature
+   * might be changed in backward-incompatible ways and is not recommended for
+   * production use. It is not subject to any SLA or deprecation policy.
+   * Creates an app profile within an instance.
+   * 
+   * @param request a {@link com.google.bigtable.admin.v2.CreateAppProfileRequest} object.
+   * @return a {@link com.google.bigtable.admin.v2.AppProfile} object.
+   */
+  AppProfile createAppProfile(CreateAppProfileRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable replication. This feature
+   * is not currently available to most Cloud Bigtable customers. This feature
+   * might be changed in backward-incompatible ways and is not recommended for
+   * production use. It is not subject to any SLA or deprecation policy.
+   * Gets information about an app profile.
+   *  
+   * @param request a {@link com.google.bigtable.admin.v2.GetAppProfileRequest} object.
+   * @return a {@link com.google.bigtable.admin.v2.AppProfile} object.
+   */
+  AppProfile getAppProfile(GetAppProfileRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable replication. This feature
+   * is not currently available to most Cloud Bigtable customers. This feature
+   * might be changed in backward-incompatible ways and is not recommended for
+   * production use. It is not subject to any SLA or deprecation policy.
+   * Lists information about app profiles in an instance.
+   * 
+   * @param request a {@link com.google.bigtable.admin.v2.ListAppProfilesRequest} object. 
+   * @return a {@link com.google.bigtable.admin.v2.ListAppProfilesResponse} object.
+   */
+  ListAppProfilesResponse listAppProfiles(ListAppProfilesRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable replication. This feature
+   * is not currently available to most Cloud Bigtable customers. This feature
+   * might be changed in backward-incompatible ways and is not recommended for
+   * production use. It is not subject to any SLA or deprecation policy.
+   * Updates an app profile within an instance.
+   * 
+   * @param request a {@link com.google.longrunning.Operation} object.
+   * @return a {@link com.google.bigtable.admin.v2.UpdateAppProfileRequest} object.
+   */
+  Operation updateAppProfile(UpdateAppProfileRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable replication. This feature
+   * is not currently available to most Cloud Bigtable customers. This feature
+   * might be changed in backward-incompatible ways and is not recommended for
+   * production use. It is not subject to any SLA or deprecation policy.
+   * Deletes an app profile from an instance.
+   * 
+   * @param request a {@link com.google.bigtable.admin.v2.DeleteAppProfileRequest} object. 
+   * @return a {@link com.google.protobuf.Empty} object.
+   */
+  Empty deleteAppProfile(DeleteAppProfileRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable instance level
+   * permissions. This feature is not currently available to most Cloud Bigtable
+   * customers. This feature might be changed in backward-incompatible ways and
+   * is not recommended for production use. It is not subject to any SLA or
+   * deprecation policy.
+   * Gets the access control policy for an instance resource. Returns an empty
+   * policy if an instance exists but does not have a policy set.
+   * 
+   * @param request a {@link com.google.iam.v1.GetIamPolicyRequest} object.
+   * @return a {@link com.google.iam.v1.Policy} object.
+   */
+  Policy getIamPolicy(GetIamPolicyRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable instance level
+   * permissions. This feature is not currently available to most Cloud Bigtable
+   * customers. This feature might be changed in backward-incompatible ways and
+   * is not recommended for production use. It is not subject to any SLA or
+   * deprecation policy.
+   * Sets the access control policy on an instance resource. Replaces any
+   * existing policy.
+   * 
+   * @param request a {@link com.google.iam.v1.SetIamPolicyRequest} object.
+   * @return a {@link com.google.iam.v1.Policy} object.
+   */
+  Policy setIamPolicy(SetIamPolicyRequest request);
+
+  /**
+   * This is a private alpha release of Cloud Bigtable instance level
+   * permissions. This feature is not currently available to most Cloud Bigtable
+   * customers. This feature might be changed in backward-incompatible ways and
+   * is not recommended for production use. It is not subject to any SLA or
+   * deprecation policy.
+   * Returns permissions that the caller has on the specified instance resource.
+   * 
+   * @param request a {@link com.google.iam.v1.TestIamPermissionsRequest} object.
+   * @return a {@link com.google.iam.v1.TestIamPermissionsResponse} object.
+   */
+  TestIamPermissionsResponse testIamPermissions(TestIamPermissionsRequest request);
 }
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 bcd2cd6a8..b67d746d1 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,19 +15,37 @@
  */
 package com.google.cloud.bigtable.grpc;
 
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
 import com.google.api.client.util.BackOff;
 import com.google.api.client.util.ExponentialBackOff;
+import com.google.bigtable.admin.v2.AppProfile;
 import com.google.bigtable.admin.v2.BigtableInstanceAdminGrpc;
 import com.google.bigtable.admin.v2.Cluster;
+import com.google.bigtable.admin.v2.CreateAppProfileRequest;
 import com.google.bigtable.admin.v2.CreateInstanceRequest;
+import com.google.bigtable.admin.v2.DeleteAppProfileRequest;
+import com.google.bigtable.admin.v2.DeleteClusterRequest;
 import com.google.bigtable.admin.v2.DeleteInstanceRequest;
+import com.google.bigtable.admin.v2.GetAppProfileRequest;
 import com.google.bigtable.admin.v2.GetClusterRequest;
 import com.google.bigtable.admin.v2.Instance;
+import com.google.bigtable.admin.v2.ListAppProfilesRequest;
+import com.google.bigtable.admin.v2.ListAppProfilesResponse;
 import com.google.bigtable.admin.v2.ListClustersRequest;
 import com.google.bigtable.admin.v2.ListClustersResponse;
 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.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.longrunning.OperationsGrpc;
@@ -35,9 +53,6 @@ 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;
 
 /**
  * <p>BigtableInstanceGrpcClient class.</p>
@@ -165,4 +180,64 @@ public class BigtableInstanceGrpcClient implements BigtableInstanceClient {
   public Operation updateCluster(Cluster cluster) {
     return instanceClient.updateCluster(cluster);
   }
+  
+  /** {@inheritDoc} */
+  @Override
+  public Empty deleteCluster(DeleteClusterRequest request) {
+    return instanceClient.deleteCluster(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public Operation partialUpdateInstance(PartialUpdateInstanceRequest request) {
+      return instanceClient.partialUpdateInstance(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public AppProfile createAppProfile(CreateAppProfileRequest request) {
+      return instanceClient.createAppProfile(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public AppProfile getAppProfile(GetAppProfileRequest request) {
+      return instanceClient.getAppProfile(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public ListAppProfilesResponse listAppProfiles(ListAppProfilesRequest request) {
+      return instanceClient.listAppProfiles(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public Operation updateAppProfile(UpdateAppProfileRequest request) {
+      return instanceClient.updateAppProfile(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public Empty deleteAppProfile(DeleteAppProfileRequest request) {
+      return instanceClient.deleteAppProfile(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public Policy getIamPolicy(GetIamPolicyRequest request) {
+      return instanceClient.getIamPolicy(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public com.google.iam.v1.Policy setIamPolicy(SetIamPolicyRequest request) {
+      return instanceClient.setIamPolicy(request);
+  }
+  
+  /** {@inheritDoc} */
+  @Override
+  public TestIamPermissionsResponse testIamPermissions(TestIamPermissionsRequest request) {
+      return instanceClient.testIamPermissions(request);
+  }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
index 1243178da..4f03dd121 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/AbstractRetryingOperation.java
@@ -135,8 +135,8 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
   protected int failedCount = 0;
 
   protected final GrpcFuture<ResultT> completionFuture;
-  protected Object callLock = new String("");
-  protected ClientCall<RequestT, ResponseT> call = NULL_CALL;
+  private Object callLock = new String("");
+  private ClientCall<RequestT, ResponseT> call = NULL_CALL;
   protected Timer.Context operationTimerContext;
   protected Timer.Context rpcTimerContext;
 
@@ -314,6 +314,10 @@ public abstract class AbstractRetryingOperation<RequestT, ResponseT, ResultT>
     }
   }
 
+  protected ClientCall<RequestT, ResponseT> getCall() {
+    return call;
+  }
+
   protected CallOptions getCallOptions() {
     if (callOptions.getDeadline() != null) {
       return callOptions;
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java
index ac59ceb87..93581c29e 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/async/RetryingStreamOperation.java
@@ -62,7 +62,7 @@ public class RetryingStreamOperation<RequestT, ResponseT>
   /** {@inheritDoc} */
   @Override
   public void onMessage(ResponseT message) {
-    call.request(1);
+    getCall().request(1);
     buffer.add(message);
   }
 
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 bc32e9bfd..27f20e321 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
@@ -66,6 +66,14 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
   public static int TIMEOUT_SECONDS = 15;
 
   private static final Logger LOG = new Logger(RefreshingOAuth2CredentialsInterceptor.class);
+  private static final HeaderCacheElement EMPTY_HEADER = new HeaderCacheElement(null, 0);
+
+  @VisibleForTesting
+  static final Metadata.Key<String> AUTHORIZATION_HEADER_KEY = Metadata.Key.of(
+      "Authorization", Metadata.ASCII_STRING_MARSHALLER);
+
+  @VisibleForTesting
+  static Clock clock = Clock.SYSTEM;
 
   /**
    * <p>
@@ -93,12 +101,6 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
     }
   }
 
-  private static final Metadata.Key<String> AUTHORIZATION_HEADER_KEY = Metadata.Key.of(
-      "Authorization", Metadata.ASCII_STRING_MARSHALLER);
-
-  @VisibleForTesting
-  static Clock clock = Clock.SYSTEM;
-
   @VisibleForTesting
   static class HeaderCacheElement {
 
@@ -119,7 +121,7 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
     final String header;
     final long actualExpirationTimeMs;
 
-    HeaderCacheElement(AccessToken token) {
+    private HeaderCacheElement(AccessToken token) {
       this.status = Status.OK;
       if (token.getExpirationTime() == null) {
         actualExpirationTimeMs = Long.MAX_VALUE;
@@ -129,13 +131,13 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
       header = "Bearer " + token.getTokenValue();
     }
 
-    HeaderCacheElement(String header, long actualExpirationTimeMs) {
+    private HeaderCacheElement(String header, long actualExpirationTimeMs) {
       this.status = Status.OK;
       this.header = header;
       this.actualExpirationTimeMs = actualExpirationTimeMs;
     }
 
-    HeaderCacheElement(Status errorStatus) {
+    private HeaderCacheElement(Status errorStatus) {
       Preconditions.checkArgument(!errorStatus.isOk(), "Error status can't be OK");
       this.status = errorStatus;
       this.header = null;
@@ -156,12 +158,31 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
       }
     }
 
-    boolean isValid() {
+    private boolean isValid() {
       return getCacheState().isValid();
     }
   }
 
-  private static final HeaderCacheElement EMPTY_HEADER = new HeaderCacheElement(null, 0);
+  private static class UnAuthResponseListener<RespT> extends SimpleForwardingClientCallListener<RespT> {
+
+    private final HeaderCacheElement origToken;
+    private final RefreshingOAuth2CredentialsInterceptor interceptor;
+
+    private UnAuthResponseListener(RefreshingOAuth2CredentialsInterceptor interceptor, Listener<RespT> delegate,
+                                   HeaderCacheElement origToken) {
+      super(delegate);
+      this.origToken = origToken;
+      this.interceptor = interceptor;
+    }
+
+    @Override
+    public void onClose(Status status, Metadata trailers) {
+      if (status == Status.UNAUTHENTICATED) {
+        interceptor.revokeUnauthToken(origToken);
+      }
+      super.onClose(status, trailers);
+    }
+  }
 
   private final ExecutorService executor;
   private final RateLimiter rateLimiter;
@@ -172,10 +193,8 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
   private Future<HeaderCacheElement> futureToken = null;
 
   // Note that the cache is volatile to allow us to peek for a Good value
-  @VisibleForTesting
   @GuardedBy("lock")
-  volatile HeaderCacheElement headerCache = EMPTY_HEADER;
-
+  private volatile HeaderCacheElement headerCache = EMPTY_HEADER;
 
   /**
    * <p>Constructor for RefreshingOAuth2CredentialsInterceptor.</p>
@@ -218,7 +237,8 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
 
         headers.put(AUTHORIZATION_HEADER_KEY, headerCache.header);
 
-        delegate().start(new UnAuthResponseListener<>(responseListener, headerCache), headers);
+        delegate().start(new UnAuthResponseListener<>(RefreshingOAuth2CredentialsInterceptor.this,
+            responseListener, headerCache), headers);
       }
 
       @Override
@@ -251,6 +271,11 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
     };
   }
 
+  @VisibleForTesting
+  CacheState getCacheState() {
+    return headerCache.getCacheState();
+  }
+
   @VisibleForTesting
   HeaderCacheElement getHeaderSafe() {
     try {
@@ -268,9 +293,7 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
   /**
    * Get the http credential header we need from a new oauth2 AccessToken.
    */
-  @VisibleForTesting
-  HeaderCacheElement getHeader() {
-
+  private HeaderCacheElement getHeader() {
     // Optimize for the common case: do a volatile read to peek for a Good cache value
     HeaderCacheElement headerCacheUnsync = this.headerCache;
 
@@ -299,7 +322,7 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
    * been refreshed.
    * This method should not be called while holding the refresh lock
    */
-  HeaderCacheElement syncRefresh() {
+  private HeaderCacheElement syncRefresh() {
     try (Closeable ss = Tracing.getTracer().spanBuilder("CredentialsRefresh").startScopedSpan()) {
       return asyncRefresh().get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
     } catch (InterruptedException e) {
@@ -399,7 +422,6 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
     }
   }
 
-
   private void revokeUnauthToken(HeaderCacheElement oldToken) {
     if (!rateLimiter.tryAcquire()) {
       LOG.trace("Rate limited");
@@ -416,24 +438,6 @@ public class RefreshingOAuth2CredentialsInterceptor implements ClientInterceptor
     }
   }
 
-  class UnAuthResponseListener<RespT> extends SimpleForwardingClientCallListener<RespT> {
-
-    private final HeaderCacheElement origToken;
-
-    UnAuthResponseListener(Listener<RespT> delegate, HeaderCacheElement origToken) {
-      super(delegate);
-      this.origToken = origToken;
-    }
-
-    @Override
-    public void onClose(Status status, Metadata trailers) {
-      if (status == Status.UNAUTHENTICATED) {
-        revokeUnauthToken(origToken);
-      }
-      super.onClose(status, trailers);
-    }
-  }
-
   @VisibleForTesting
   boolean isRefreshing() {
     synchronized(lock) {
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
index 9f97f875a..2aa3fd567 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RetryingReadRowsOperation.java
@@ -57,22 +57,22 @@ public class RetryingReadRowsOperation extends
 
     @Override
     public void onNext(ReadRowsRequest value) {
-      call.sendMessage(value);
+      getCall().sendMessage(value);
     }
 
     @Override
     public void onError(Throwable t) {
-      call.cancel("Cancelled by client with StreamObserver.onError()", t);
+      getCall().cancel("Cancelled by client with StreamObserver.onError()", t);
     }
 
     @Override
     public void onCompleted() {
-      call.halfClose();
+      getCall().halfClose();
     }
 
     @Override
     public boolean isReady() {
-      return call.isReady();
+      return getCall().isReady();
     }
 
     @Override
@@ -86,17 +86,17 @@ public class RetryingReadRowsOperation extends
 
     @Override
     public void request(int count) {
-      call.request(count);
+      getCall().request(count);
     }
 
     @Override
     public void setMessageCompression(boolean enable) {
-      call.setMessageCompression(enable);
+      getCall().setMessageCompression(enable);
     }
 
     @Override
     public void cancel(@Nullable String s, @Nullable Throwable throwable) {
-      call.cancel(s, throwable);
+      getCall().cancel(s, throwable);
     }
   }
 
@@ -148,15 +148,13 @@ public class RetryingReadRowsOperation extends
   public void run() {
     try {
       // restart the clock.
-      synchronized (callLock) {
-        super.run();
-        // pre-fetch one more result, for performance reasons.
-        adapter.request(1);
-        if (rowObserver instanceof ClientResponseObserver) {
-          ((ClientResponseObserver<ReadRowsRequest, FlatRow>) rowObserver).beforeStart(adapter);
-        }
-        lastResponseMs = clock.currentTimeMillis();
+      super.run();
+      // pre-fetch one more result, for performance reasons.
+      adapter.request(1);
+      if (rowObserver instanceof ClientResponseObserver) {
+        ((ClientResponseObserver<ReadRowsRequest, FlatRow>) rowObserver).beforeStart(adapter);
       }
+      lastResponseMs = clock.currentTimeMillis();
     } catch (Exception e) {
       setException(e);
     }
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java
index 26da94986..4408b4f5c 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/ExportJob.java
@@ -17,16 +17,17 @@ package com.google.cloud.bigtable.beam.sequencefiles;
 
 import com.google.cloud.bigtable.beam.CloudBigtableIO;
 import com.google.cloud.bigtable.beam.CloudBigtableScanConfiguration;
-import com.google.cloud.bigtable.beam.sequencefiles.Utils.StringToDirectoryResourceId;
+import java.io.Serializable;
 import java.nio.charset.CharacterCodingException;
 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.DefaultFilenamePolicy;
-import org.apache.beam.sdk.io.LocalResources;
+import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.io.Read;
 import org.apache.beam.sdk.io.WriteFiles;
+import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
 import org.apache.beam.sdk.io.fs.ResourceId;
 import org.apache.beam.sdk.options.Default;
 import org.apache.beam.sdk.options.Description;
@@ -34,6 +35,7 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.options.ValueProvider;
 import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
 import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.transforms.SimpleFunction;
 import org.apache.beam.sdk.values.KV;
 import org.apache.commons.logging.Log;
@@ -125,7 +127,7 @@ public class ExportJob {
     void setBigtableFilter(String filter);
 
 
-    @Description("The destination path. Can either specify a directory ending with a / or a file prefix.")
+    @Description("The destination directory")
     ValueProvider<String> getDestinationPath();
     @SuppressWarnings("unused")
     void setDestinationPath(ValueProvider<String> destinationPath);
@@ -189,15 +191,17 @@ public class ExportJob {
       configBuilder.withAppProfileId(opts.getBigtableAppProfileId());
     }
 
+    // Use the base target directory to stage bundles
+    ValueProvider<ResourceId> destinationPath = NestedValueProvider
+        .of(opts.getDestinationPath(), new StringToDirResourceId());
 
-    ValueProvider<ResourceId> dest = NestedValueProvider.of(
-        opts.getDestinationPath(), new StringToDirectoryResourceId()
-    );
+    // Concat the destination path & prefix for the final path
+    FilePathPrefix filePathPrefix = new FilePathPrefix(destinationPath, opts.getFilenamePrefix());
 
     SequenceFileSink<ImmutableBytesWritable, Result> sink = new SequenceFileSink<>(
-        dest,
-        DefaultFilenamePolicy.constructUsingStandardParameters(
-            LocalResources.fromString(opts.getFilenamePrefix(), false),
+        destinationPath,
+        DefaultFilenamePolicy.fromStandardParameters(
+            filePathPrefix,
             null,
             "",
             false
@@ -223,4 +227,31 @@ public class ExportJob {
     }
   }
 
+  static class StringToDirResourceId implements SerializableFunction<String, ResourceId>, Serializable {
+    @Override
+    public ResourceId apply(String input) {
+      return FileSystems.matchNewResource(input, true);
+    }
+  }
+
+  static class FilePathPrefix implements ValueProvider<ResourceId>, Serializable {
+    private final ValueProvider<ResourceId> destinationPath;
+    private final ValueProvider<String> filenamePrefix;
+
+    FilePathPrefix(ValueProvider<ResourceId> destinationPath,
+        ValueProvider<String> filenamePrefix) {
+      this.destinationPath = destinationPath;
+      this.filenamePrefix = filenamePrefix;
+    }
+
+    @Override
+    public ResourceId get() {
+      return destinationPath.get().resolve(filenamePrefix.get(), StandardResolveOptions.RESOLVE_FILE);
+    }
+
+    @Override
+    public boolean isAccessible() {
+      return destinationPath.isAccessible() && filenamePrefix.isAccessible();
+    }
+  }
 }
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSink.java b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSink.java
index 9eaaf6d35..4f1d62520 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSink.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/main/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSink.java
@@ -22,6 +22,8 @@ import java.nio.ByteBuffer;
 import java.nio.channels.WritableByteChannel;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
+import org.apache.beam.sdk.io.Compression;
+import org.apache.beam.sdk.io.DynamicFileDestinations;
 import org.apache.beam.sdk.io.FileBasedSink;
 import org.apache.beam.sdk.io.fs.ResourceId;
 import org.apache.beam.sdk.options.ValueProvider;
@@ -41,7 +43,7 @@ import org.apache.hadoop.io.serializer.Serialization;
  * @param <K> The type of the {@link SequenceFile} key.
  * @param <V> The type of the {@link SequenceFile} value.
  */
-class SequenceFileSink<K,V> extends FileBasedSink<KV<K, V>> {
+class SequenceFileSink<K,V> extends FileBasedSink<KV<K,V>, Void, KV<K, V>> {
   private static final Log LOG = LogFactory.getLog(SequenceFileSink.class);
 
   private final Class<K> keyClass;
@@ -56,7 +58,7 @@ class SequenceFileSink<K,V> extends FileBasedSink<KV<K, V>> {
       FilenamePolicy filenamePolicy,
       Class<K> keyClass, Class<? extends Serialization<? super K>> keySerializationClass,
       Class<V> valueClass, Class<? extends Serialization<? super V>> valueSerializationClass) {
-    super(baseOutputDirectoryProvider, filenamePolicy, CompressionType.UNCOMPRESSED);
+    super(baseOutputDirectoryProvider, DynamicFileDestinations.<KV<K,V>>constant(filenamePolicy), Compression.UNCOMPRESSED);
 
     this.keyClass = keyClass;
     this.valueClass = valueClass;
@@ -72,7 +74,7 @@ class SequenceFileSink<K,V> extends FileBasedSink<KV<K, V>> {
    * {@inheritDoc}
    */
   @Override
-  public WriteOperation<KV<K, V>> createWriteOperation() {
+  public WriteOperation<Void, KV<K, V>> createWriteOperation() {
     return new SeqFileWriteOperation<>(this, keyClass, valueClass, serializationNames);
   }
 
@@ -84,7 +86,7 @@ class SequenceFileSink<K,V> extends FileBasedSink<KV<K, V>> {
    * @param <K> The type of the {@link SequenceFile} key.
    * @param <V> The type of the {@link SequenceFile} value.
    */
-  private static class SeqFileWriteOperation<K,V> extends WriteOperation<KV<K,V>> {
+  private static class SeqFileWriteOperation<K,V> extends WriteOperation<Void, KV<K,V>> {
     private final Class<K> keyClass;
     private final Class<V> valueClass;
     private final String[] serializationNames;
@@ -114,7 +116,7 @@ class SequenceFileSink<K,V> extends FileBasedSink<KV<K, V>> {
      * {@inheritDoc}
      */
     @Override
-    public Writer<KV<K, V>> createWriter() throws Exception {
+    public Writer<Void, KV<K, V>> createWriter() throws Exception {
       return new SeqFileWriter<>(this);
     }
   }
@@ -126,7 +128,7 @@ class SequenceFileSink<K,V> extends FileBasedSink<KV<K, V>> {
    * @param <K> The type of the {@link SequenceFile} key.
    * @param <V> The type of the {@link SequenceFile} value.
    */
-  private static class SeqFileWriter<K,V> extends Writer<KV<K,V>> {
+  private static class SeqFileWriter<K,V> extends Writer<Void, KV<K,V>> {
     private final SeqFileWriteOperation<K,V> writeOperation;
     private SequenceFile.Writer sequenceFile;
     private final AtomicLong counter = new AtomicLong();
diff --git a/pom.xml b/pom.xml
index 0a0c38da8..f1719fce9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -67,7 +67,7 @@ limitations under the License.
         <dropwizard.metrics.version>3.1.2</dropwizard.metrics.version>
 
 
-        <beam.version>2.1.0</beam.version>
+        <beam.version>2.3.0</beam.version>
         <commons-lang.version>2.6</commons-lang.version>
         <google-auto-service.version>1.0-rc2</google-auto-service.version>
         <auto-value.version>1.1</auto-value.version>


Test patch:
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 7553c81d5..d4a29c820 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
@@ -17,16 +17,23 @@ package com.google.cloud.bigtable.grpc.io;
 
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
 import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.api.client.util.Clock;
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.ReadRowsRequest;
+import com.google.bigtable.v2.ReadRowsResponse;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.auth.oauth2.AccessToken;
 import com.google.auth.oauth2.OAuth2Credentials;
 import com.google.cloud.bigtable.grpc.io.RefreshingOAuth2CredentialsInterceptor.CacheState;
 import com.google.cloud.bigtable.grpc.io.RefreshingOAuth2CredentialsInterceptor.HeaderCacheElement;
+
 import java.io.IOException;
 import java.util.Date;
 import java.util.concurrent.Callable;
@@ -36,6 +43,8 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+
+import io.grpc.*;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
@@ -43,6 +52,7 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
@@ -50,6 +60,9 @@ import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 
+/**
+ * Tests for {@link RefreshingOAuth2CredentialsInterceptor}
+ */
 @RunWith(JUnit4.class)
 public class RefreshingOAuth2CredentialsInterceptorTest {
 
@@ -68,12 +81,23 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
   private RefreshingOAuth2CredentialsInterceptor underTest;
 
   @Mock
-  private OAuth2Credentials credentials;
+  private OAuth2Credentials mockCredentials;
+
+  @Mock
+  private Channel mockChannel;
+
+  @Mock
+  private ClientCall mockClientCall;
+
+  @Mock
+  private ClientCall.Listener mockListener;
 
   @Before
   public void setupMocks() {
     MockitoAnnotations.initMocks(this);
     setTimeInMillieconds(0L);
+    when(mockChannel.newCall(any(MethodDescriptor.class), any(CallOptions.class)))
+        .thenReturn(mockClientCall);
   }
 
   private void setTimeInMillieconds(final long timeMs) {
@@ -87,19 +111,70 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
 
   @Test
   public void testSyncRefresh() throws IOException {
-    initialize(HeaderCacheElement.TOKEN_STALENESS_MS + 1);
-    Assert.assertEquals(CacheState.Good, underTest.headerCache.getCacheState());
+    Mockito.when(mockCredentials.refreshAccessToken()).thenReturn(
+        new AccessToken("", new Date(HeaderCacheElement.TOKEN_STALENESS_MS + 1)));
+    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, mockCredentials);
+    Assert.assertEquals(CacheState.Expired, underTest.getCacheState());
+    underTest.getHeaderSafe();
+    Assert.assertNotEquals(CacheState.Exception, underTest.getCacheState());
+    Assert.assertEquals(CacheState.Good, underTest.getCacheState());
     Assert.assertFalse(underTest.isRefreshing());
   }
 
+  @Test
+  /**
+   * Basic test to make sure that the interceptor works properly
+   */
+  public void testIntercept() throws IOException {
+    initialize(HeaderCacheElement.TOKEN_STALENESS_MS + 1);
+    initializeCall(CallOptions.DEFAULT);
+  }
+
+  @Test
+  /**
+   * Test to make sure that the interceptor revokes credentials if an RPC receives an UNAUTHENTICATED status.
+   */
+  public void testInvalid() throws IOException {
+    long expiration = HeaderCacheElement.TOKEN_STALENESS_MS + 1;
+    initialize(expiration);
+    ClientCall.Listener listener = initializeCall(CallOptions.DEFAULT);
+    listener.onClose(Status.UNAUTHENTICATED, new Metadata());
+    Assert.assertEquals(CacheState.Expired, underTest.getCacheState());
+  }
+
+  private ClientCall.Listener initializeCall(CallOptions callOptions) throws IOException {
+    ClientCall<ReadRowsRequest, ReadRowsResponse> call = underTest.interceptCall(
+        BigtableGrpc.METHOD_READ_ROWS,
+        callOptions,
+        mockChannel);
+    Metadata metadata = new Metadata();
+    call.start(mockListener, metadata);
+    Assert.assertEquals(underTest.getHeaderSafe().header,
+        metadata.get(RefreshingOAuth2CredentialsInterceptor.AUTHORIZATION_HEADER_KEY));
+
+    ArgumentCaptor<ClientCall.Listener> listenerCaptor = ArgumentCaptor.forClass(ClientCall.Listener.class);
+    verify(mockClientCall, times(1))
+        .start(listenerCaptor.capture(), same(metadata));
+
+    call.request(1);
+    call.sendMessage(ReadRowsRequest.getDefaultInstance());
+
+    verify(mockClientCall, times(1))
+        .request(eq(1));
+    verify(mockClientCall, times(1))
+        .sendMessage(same(ReadRowsRequest.getDefaultInstance()));
+
+    return listenerCaptor.getValue();
+  }
+
   @Test
   public void testRefresh() throws IOException {
-    Mockito.when(credentials.refreshAccessToken()).thenReturn(
+    Mockito.when(mockCredentials.refreshAccessToken()).thenReturn(
         new AccessToken("", new Date(HeaderCacheElement.TOKEN_STALENESS_MS + 1)));
     underTest = new RefreshingOAuth2CredentialsInterceptor(MoreExecutors.newDirectExecutorService(),
-        credentials);
-    underTest.syncRefresh();
-    Assert.assertEquals(CacheState.Good, underTest.headerCache.getCacheState());
+        mockCredentials);
+    underTest.getHeaderSafe();
+    Assert.assertEquals(CacheState.Good, underTest.getCacheState());
     Assert.assertFalse(underTest.isRefreshing());
   }
 
@@ -108,7 +183,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
   public void testExecutorProblem() throws IOException {
     ExecutorService mockExecutor = Mockito.mock(ExecutorService.class);
     when(mockExecutor.submit(any(Callable.class))).thenThrow(new RuntimeException(""));
-    underTest = new RefreshingOAuth2CredentialsInterceptor(mockExecutor, credentials);
+    underTest = new RefreshingOAuth2CredentialsInterceptor(mockExecutor, mockCredentials);
     Assert.assertEquals(CacheState.Exception, underTest.getHeaderSafe().getCacheState());
   }
 
@@ -116,35 +191,38 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
   public void testStaleAndExpired() throws IOException {
     long expiration = HeaderCacheElement.TOKEN_STALENESS_MS + 1;
     initialize(expiration);
-    Assert.assertEquals(CacheState.Good, underTest.headerCache.getCacheState());
+    Assert.assertEquals(CacheState.Good, underTest.getCacheState());
     long startTime = 2L;
     setTimeInMillieconds(startTime);
-    Assert.assertEquals(CacheState.Stale, underTest.headerCache.getCacheState());
+    Assert.assertEquals(CacheState.Stale, underTest.getCacheState());
     long expiredStaleDiff =
         HeaderCacheElement.TOKEN_STALENESS_MS - HeaderCacheElement.TOKEN_EXPIRES_MS;
     setTimeInMillieconds(startTime + expiredStaleDiff);
-    Assert.assertEquals(CacheState.Expired, underTest.headerCache.getCacheState());
+    Assert.assertEquals(CacheState.Expired, underTest.getCacheState());
   }
 
   @Test
-  public void testNullExpiration() {
+  public void testNullExpiration() throws Exception {
     setTimeInMillieconds(100);
-    HeaderCacheElement element = new HeaderCacheElement(new AccessToken("", null));
-    Assert.assertEquals(CacheState.Good, element.getCacheState());
+    Mockito.when(mockCredentials.refreshAccessToken()).thenReturn(
+        new AccessToken("", null));
+    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, mockCredentials);
+    underTest.asyncRefresh().get(100, TimeUnit.MILLISECONDS);
+    Assert.assertEquals(CacheState.Good, underTest.getCacheState());
 
     // Make sure that the header doesn't expire in the distant future.
     setTimeInMillieconds(100000000000L);
-    Assert.assertEquals(CacheState.Good, element.getCacheState());
+    Assert.assertEquals(CacheState.Good, underTest.getCacheState());
   }
 
   @Test
   public void testRefreshAfterFailure() throws Exception {
-    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, credentials);
+    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, mockCredentials);
 
     final AccessToken accessToken = new AccessToken("hi", new Date(HeaderCacheElement.TOKEN_STALENESS_MS + 1));
 
     //noinspection unchecked
-    Mockito.when(credentials.refreshAccessToken())
+    Mockito.when(mockCredentials.refreshAccessToken())
         // First call will throw Exception & bypass retries
         .thenThrow(new IOException())
         // Second call will succeed
@@ -159,12 +237,12 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     Assert.assertEquals(CacheState.Good, secondResult.getCacheState());
     Assert.assertThat(secondResult.header, containsString("hi"));
     // Make sure that the token was only requested twice: once for the first failure & second time for background recovery
-    Mockito.verify(credentials, times(2)).refreshAccessToken();
+    Mockito.verify(mockCredentials, times(2)).refreshAccessToken();
   }
 
   @Test
   public void testRefreshAfterStale() throws Exception {
-    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, credentials);
+    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, mockCredentials);
 
     long expires = HeaderCacheElement.TOKEN_STALENESS_MS + 1;
     final AccessToken staleToken = new AccessToken("stale", new Date(expires));
@@ -172,7 +250,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
         new AccessToken("good", new Date(expires + HeaderCacheElement.TOKEN_STALENESS_MS + 1));
 
     //noinspection unchecked
-    Mockito.when(credentials.refreshAccessToken())
+    Mockito.when(mockCredentials.refreshAccessToken())
         // First call will setup a stale token
         .thenReturn(staleToken)
         // Second call will give a good token
@@ -200,10 +278,10 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     Assert.assertThat(thirdResult.header, containsString("good"));
 
     // Make sure that the token was only requested twice: once for the stale token & second time for the good token
-    Mockito.verify(credentials, times(2)).refreshAccessToken();
+    Mockito.verify(mockCredentials, times(2)).refreshAccessToken();
   }
 
-  @Test(timeout=30000)
+  @Test(timeout = 30000)
   /*
    * Test that checks that concurrent requests to RefreshingOAuth2CredentialsInterceptor refresh
    * logic doesn't cause hanging behavior.  Specifically, when an Expired condition occurs it
@@ -230,22 +308,22 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     final long expiration = start + HeaderCacheElement.TOKEN_EXPIRES_MS + 1;
 
     // Create a mechanism that will allow us to control when the accessToken is returned.
-    // credentials.refreshAccessToken() will get called asynchronously and will wait until the
+    // mockCredentials.refreshAccessToken() will get called asynchronously and will wait until the
     // lock is notified before returning.  That will allow us to set up multiple concurrent calls
     final FutureAnswer<AccessToken> answer = new FutureAnswer<>();
-    Mockito.when(credentials.refreshAccessToken()).thenAnswer(answer);
+    Mockito.when(mockCredentials.refreshAccessToken()).thenAnswer(answer);
 
     underTest =
-        new RefreshingOAuth2CredentialsInterceptor(executorService, credentials);
+        new RefreshingOAuth2CredentialsInterceptor(executorService, mockCredentials);
 
     // At this point, the access token wasn't retrieved yet. The
     // RefreshingOAuth2CredentialsInterceptor considers null to be Expired.
-    Assert.assertEquals(CacheState.Expired, underTest.headerCache.getCacheState());
+    Assert.assertEquals(CacheState.Expired, underTest.getCacheState());
 
     syncCall(answer, new Date(expiration));
 
     // Check to make sure that the AccessToken was retrieved.
-    Assert.assertEquals(CacheState.Stale, underTest.headerCache.getCacheState());
+    Assert.assertEquals(CacheState.Stale, underTest.getCacheState());
 
     // Check to make sure we're no longer refreshing.
     Assert.assertFalse(underTest.isRefreshing());
@@ -256,7 +334,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     // necessary, but also should not be harmful, since there are likely to be multiple concurrent
     // requests that call asyncRefresh() when the token turns stale.
     Future<HeaderCacheElement> previous = underTest.asyncRefresh();
-    for (int i = 0; i < 10; i++) {
+    for (int i = 0; i < 100; i++) {
       Future<HeaderCacheElement> current = underTest.asyncRefresh();
       Assert.assertEquals(previous, current);
       previous = current;
@@ -266,7 +344,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     Assert.assertFalse(underTest.isRefreshing());
   }
 
-  private static class FutureAnswer<T> implements Answer<T>{
+  private static class FutureAnswer<T> implements Answer<T> {
 
     private SettableFuture<T> future = SettableFuture.create();
 
@@ -280,7 +358,7 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     }
 
     void reset() {
-      future = SettableFuture.create();;
+      future = SettableFuture.create();
     }
   }
 
@@ -295,16 +373,13 @@ public class RefreshingOAuth2CredentialsInterceptorTest {
     Assert.assertTrue(underTest.isRefreshing());
 
     answer.set(new AccessToken("hi", expirationTime));
-    // Wait for no more than a second to make sure that the call to underTest.syncRefresh()
-    // completes properly.  If a second passes without syncRefresh() completing, future.get(..)
-    // will throw a TimeoutException.
     underTest.asyncRefresh().get(100, TimeUnit.MILLISECONDS);
   }
 
   private void initialize(long expiration) throws IOException {
-    Mockito.when(credentials.refreshAccessToken()).thenReturn(
+    Mockito.when(mockCredentials.refreshAccessToken()).thenReturn(
         new AccessToken("", new Date(expiration)));
-    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, credentials);
-    underTest.syncRefresh();
+    underTest = new RefreshingOAuth2CredentialsInterceptor(executorService, mockCredentials);
+    underTest.getHeaderSafe();
   }
 }
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSinkTest.java b/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSinkTest.java
index 8da305519..c007b31bc 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSinkTest.java
+++ b/bigtable-dataflow-parent/bigtable-beam-import/src/test/java/com/google/cloud/bigtable/beam/sequencefiles/SequenceFileSinkTest.java
@@ -61,7 +61,7 @@ public class SequenceFileSinkTest {
     );
 
     FilenamePolicy filenamePolicy = DefaultFilenamePolicy
-        .constructUsingStandardParameters(output, null, null, false);
+        .fromStandardParameters(output, null, null, false);
 
     SequenceFileSink<Text, Text> sink = new SequenceFileSink<>(
         output, filenamePolicy,