Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1723 (2018-03-11 17:26:26)

AbstractRetryingOperation's call is now private.
`call` should be read only.


Patch:
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);
     }


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();
   }
 }