Updates to the AuthenticationRefresher timer

The AuthenticationRefreher will now keep a instance-level reference to
the current Executor. A singleton style synchronization ensures there is
only ever one Executor active.

Additionally, the APIClient passed into .provide is now heald as a
WeakReference by the Executor. if the APIClient is ever garbage
collected the Executor is cancelled and nulled.

Author: raykrueger

util/src/main/java/io/kubernetes/client/util/credentials/AuthenticationRefresher.java

@@ -12,8 +12,10 @@
 */
 package io.kubernetes.client.util.credentials;
 
+import java.lang.ref.WeakReference;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,11 +28,12 @@
  *
  * Can be used with ClientBuilder as such:
  *
- * <pre>
- * ClientBuilder.standard()
- *         .withAuthentication(new AuthenticationRefresher(new KubeconfigAuthentication(), 60))
- *         .build();
- * </pre>
+ * <pre> ClientBuilder.standard() .withAuthentication(new
+ * AuthenticationRefresher(new KubeconfigAuthentication(), 60)) .build(); </pre>
+ *
+ * The AuthenticationRefresher maintains a {@link WeakReference} to the
+ * ApiClient passed into provide. When the client instance goes out of scope,
+ * the refresh timer will be cancelled and released.
  */
 public class AuthenticationRefresher implements Authentication {
 
@@ -39,9 +42,13 @@ public class AuthenticationRefresher implements Authentication {
     private final Authentication delegateAuthentication;
     private final Long expirationSeconds;
 
+    private final ScheduledExecutorService executor;
+    private ScheduledFuture<?> currentSchedule;
+
     public AuthenticationRefresher(Authentication delegateAuthentication, long expirationSeconds) {
         this.delegateAuthentication = delegateAuthentication;
         this.expirationSeconds = expirationSeconds;
+        this.executor = Executors.newSingleThreadScheduledExecutor();
         log.debug("AuthenticationRefresher initialized with expirationSeconds: " + expirationSeconds);
     }
 
@@ -50,17 +57,40 @@ public AuthenticationRefresher(Authentication delegateAuthentication, long expir
      */
     @Override
     public void provide(ApiClient client) {
-        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
-        executor.scheduleAtFixedRate(new Runnable() {
+
+        synchronized (this) {
+            if (currentSchedule == null) {
+                currentSchedule = scheduleRefresh(client);
+            }
+        }
+
+        // Run it now, synchronously.
+        log.debug("Invoking authentication");
+        this.delegateAuthentication.provide(client);
+    }
+
+    private ScheduledFuture<?> scheduleRefresh(final ApiClient clientArg) {
+        WeakReference<ApiClient> clientReference = new WeakReference<>(clientArg);
+
+        return executor.scheduleAtFixedRate(new Runnable() {
             @Override
             public void run() {
                 log.debug("Refreshing authentication");
-                delegateAuthentication.provide(client);
+
+                ApiClient client = clientReference.get();
+
+                if (client == null) {
+                    log.debug("Delegate authentication is null, stopping");
+                    currentSchedule.cancel(true);
+                    currentSchedule = null;
+                    return;
+                } else {
+                    log.debug("Invoking authentication");
+                    delegateAuthentication.provide(client);
+                }
+
             }
         }, expirationSeconds, expirationSeconds, java.util.concurrent.TimeUnit.SECONDS);
 
-        // Run it now, synchronously.
-        log.debug("Invoking authentication");
-        delegateAuthentication.provide(client);
     }
 }

util/src/test/java/io/kubernetes/client/util/ClientBuilderTest.java

@@ -16,8 +16,8 @@
 import static io.kubernetes.client.util.Config.ENV_SERVICE_PORT;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.io.File;
@@ -26,6 +26,8 @@
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -224,15 +226,20 @@ void credentialProviderInvoked() throws IOException {
 
   @Test
   void credentialProviderWrappedWithRefresher() throws IOException, InterruptedException {
-    final Authentication provider = mock(Authentication.class);
-    final ApiClient client = ClientBuilder.standard()
+    CountDownLatch latch = new CountDownLatch(2);
+    final Authentication provider = new Authentication() {
+      @Override
+      public void provide(ApiClient client) {
+        latch.countDown();
+      }
+    };
+    ClientBuilder.standard()
         .setAuthentication(provider)
         .setAuthenticationRefreshSeconds(Duration.ofSeconds(2))
         .build();
 
-    Thread.sleep(3000);
-
-    verify(provider, times(2)).provide(client);
+    boolean result = latch.await(4, TimeUnit.SECONDS);
+    assertTrue(result, "CountDownLatch timed out, which means our provider was not called twice.");
   }
 
   /**

util/src/test/java/io/kubernetes/client/util/credentials/AuthenticationRefresherTest.java

@@ -12,11 +12,15 @@
 */
 package io.kubernetes.client.util.credentials;
 
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.io.IOException;
+import java.lang.ref.WeakReference;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 
@@ -32,21 +36,63 @@ void credentialProviderWrapperInvoked() throws IOException {
 
         final ApiClient client = ClientBuilder.standard().setAuthentication(wrapper).build();
 
+        //No latch here, just need to make sure it got called once.
         verify(provider).provide(client);
     }
 
-    @Test
-    void credentialProviderTimerInvoked() throws Exception {
-        final Authentication provider = mock(Authentication.class);
+    /**
+     * CountdownLatch should be called 2 times in this teest. Once
+     * synchronously, and once asynchronously.
+     *
+     * @throws Exception
+     */
+    @Test void credentialProviderTimerInvoked() throws Exception {
+    CountDownLatch latch = new CountDownLatch(2);
+
+        final Authentication provider = new Authentication() {
+            @Override
+            public void provide(ApiClient client) {
+                latch.countDown();
+            }
+        };
         final Authentication wrapper = new AuthenticationRefresher(provider, 2);
 
-        final ApiClient client = ClientBuilder.standard().setAuthentication(wrapper).build();
+        ClientBuilder.standard().setAuthentication(wrapper).build();
+
+        boolean result = latch.await(4, TimeUnit.SECONDS);
+        assertTrue(result, "CountDownLatch timed out, which means our provider was not called twice.");
+    }
+
+    /**
+     * CountdownLatch should be called 1 time in this teest. Once
+     * synchronously, and then the timer should cancel.
+     *
+     * @throws Exception
+     */
+    @Test void credentialProviderTimerInvokedOnce() throws Exception {
+    CountDownLatch latch = new CountDownLatch(2);
+
+        final Authentication provider = new Authentication() {
+            @Override
+            public void provide(ApiClient client) {
+                latch.countDown();
+            }
+        };
+        final Authentication wrapper = new AuthenticationRefresher(provider, 2);
 
-        Thread.sleep(3000);
+        ClientBuilder.standard().setAuthentication(wrapper).build();
 
-        // verify provider mock called 2 times
-        verify(provider, times(2)).provide(client);
+        //Crazy hack to force garbage collection and wait for it
+        Object obj = new Object();
+        WeakReference<Object> ref = new WeakReference<>(obj);
+        obj = null;
+        while (ref.get() != null) {
+            System.gc();
+        }
 
+        boolean result = latch.await(4, TimeUnit.SECONDS);
+        assertFalse(result,
+                "CountDownLatch SHOULD HAVE timed out, which means the ScheduledExecutorService was not cancelled on GC.");
     }
 
 }