Repository: firebase/firebase-admin-java
Pull request #77 (2017-09-25 19:57:26)

Improving the ThreadManager and TokenRefresher APIs
This is a follow up to #74. It attempts to solve several issues in the current design:

* `ThreadManager` interface returns a `ScheduledExecutorService`. This makes implementing the `ThreadManager` API cumbersome, since the default implementation provided in Java (`ScheduledThreadPoolExecutor`) is not very amenable to configuration.
* The different uses of the `getExecutor()` and `getThreadManager()` methods are not well defined in the `ThreadManager` interface. At the moment it is something along the lines of:
  - `getThreadManager()`: for RTDB
  - `getExecutor()`: for everything else
* Proactive token refresher always get started when calling `initializeApp()` (which starts long-lived threads). But only RTDB requires it in practice. This is not ideal when using the SDK only for auth operations.

To resolve these issue, this PR makes the following changes:

* Clearly define the contract of the `ThreadManager` interface.
  - `getThreadMaanger()`: Used to start long-lived threads (including RTDB, token refresher and anything else that would require long-lived threads in the future)
  - `getExecutor()`: Used for short-lived tasks.
* Return an `ExecutorService` from `ThreadManager`, whose default implementation (`ThreadPoolExecutor`) is much easier to configure.
* Invert the control of the proactive token refresher. Instead of always starting it from the app, let RTDB (or any other interested component) start it on demand.

The downside of this change is that we end up using a dedicated executor for proactive token refresher (we can use it for other scheduled tasks that we may add in the future -- but for now it's only used for token refresher). In GAE this will end up occupying a separate background thread. This shouldn't be a problem for those who are only using RTDB, but GAE users who access both RTDB and auth ends up spending another thread.  However, it is possible to provide a custom `ThreadManager` that uses request-scoped threads for short-lived tasks, which is one way to avoid this drawback.

hiranya911: I did some testing on GAE for this PR. Auto and manual scaling works as before. Basic scaling also works, and idle instance shutdown can be easily enabled with a custom `ThreadManager`:

```
static class BasicScalingThreadManager extends ThreadManager {

    @Override
    protected ExecutorService getExecutor(@NonNull FirebaseApp firebaseApp) {
      // Use a single-threaded executor which keeps threads alive for 1 minute.
      return new ThreadPoolExecutor(0, 1,
          60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), getThreadFactory());
    }

    @Override
    protected void releaseExecutor(@NonNull FirebaseApp firebaseApp,
        @NonNull ExecutorService executorService) {
      executorService.shutdownNow();
    }

    @Override
    protected ThreadFactory getThreadFactory() {
      return com.google.appengine.api.ThreadManager.backgroundThreadFactory();
    }
}
```

My test GAE instance also has a 1 minute idle timeout. With this setup, when the app doesn't receive requests for 2 minutes (1 minute for the threads to die, and 1 more minute for the GAE idle timeout to kick in), the instance is stopped.

```
15:36:57.769  GET 404 270 B 7.2 s Unknown /_ah/start GET 404 270 B 7.2 s Unknown 
15:37:06.974  GET 200 2 B 71.6 s Unknown /_ah/background GET 200 2 B 71.6 s Unknown
15:37:16.326  GET 200 119 B 231 ms curl/7.53.1 /users?uid=jALVj6oEyvXmwLZuMcUCOeRWxhS2&count=1 GET 200 119 B 231 ms curl/7.53.1 
15:37:17.414  GET 200 119 B 230 ms curl/7.53.1 /users?uid=jALVj6oEyvXmwLZuMcUCOeRWxhS2&count=1 GET 200 119 B 230 ms curl/7.53.1
15:37:18.383  GET 200 119 B 213 ms curl/7.53.1 /users?uid=jALVj6oEyvXmwLZuMcUCOeRWxhS2&count=1 GET 200 119 B 213 ms curl/7.53.1
15:39:19.068  GET 200 2 B 93 ms Unknown /_ah/stop GET 200 2 B 93 ms Unknown
15:39:57.372  GET 200 119 B 13.3 s curl/7.53.1 /users?uid=jALVj6oEyvXmwLZuMcUCOeRWxhS2&count=1 GET 200 119 B 13.3 s curl/7.53.1
15:39:57.382  GET 404 270 B 8.3 s Unknown /_ah/start GET 404 270 B 8.3 s Unknown
15:40:06.735  GET 200 2 B 64 s Unknown /_ah/background GET 200 2 B 64 s Unknown
15:42:11.231  GET 200 2 B 93 ms Unknown /_ah/stop GET 200 2 B 93 ms Unknown 
```
googlebot: Thanks for your pull request.  It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: **Please visit <https://cla.developers.google.com/> to sign.**

Once you've signed, please reply here (e.g. `I signed it!`) and we'll verify.  Thanks.

---

- If you've already signed a CLA, it's possible we don't have your GitHub username or you're using a different email address.  Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).
- If your company signed a CLA, they designated a Point of Contact who decides which employees are authorized to participate. You may need to contact the Point of Contact for your company and ask to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the project maintainer to go/cla#troubleshoot.
- In order to pass this check, please resolve this problem and have the pull request author add another comment and the bot will run again.

<!-- need_sender_cla -->
hiranya911: I've been conducting a long running test with this PR on App Engine (manual scaling). The test is coming up on 24 hours now and seem to be working fine. 
googlebot: CLAs look good, thanks!

<!-- ok -->
hiranya911: Made some of the suggested changes. Responded to all open comments. On to the next round.

Patch:
diff --git a/src/main/java/com/google/firebase/FirebaseApp.java b/src/main/java/com/google/firebase/FirebaseApp.java
index 91c47fc..029b1ea 100644
--- a/src/main/java/com/google/firebase/FirebaseApp.java
+++ b/src/main/java/com/google/firebase/FirebaseApp.java
@@ -32,12 +32,13 @@ import com.google.common.base.MoreObjects;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.BaseEncoding;
-import com.google.common.util.concurrent.ListeningScheduledExecutorService;
 import com.google.firebase.internal.FirebaseAppStore;
 import com.google.firebase.internal.FirebaseService;
+import com.google.firebase.internal.GaeThreadFactory;
 import com.google.firebase.internal.NonNull;
 import com.google.firebase.internal.Nullable;
 
+import com.google.firebase.internal.RevivingScheduledExecutor;
 import com.google.firebase.tasks.Task;
 import com.google.firebase.tasks.Tasks;
 import java.io.IOException;
@@ -50,11 +51,13 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -90,10 +93,12 @@ public class FirebaseApp {
   private final FirebaseOptions options;
   private final TokenRefresher tokenRefresher;
   private final ThreadManager threadManager;
+  private final ThreadManager.FirebaseExecutors executors;
 
   private final AtomicBoolean deleted = new AtomicBoolean();
   private final Map<String, FirebaseService> services = new HashMap<>();
-  private final ListeningScheduledExecutorService executor;
+
+  private volatile ScheduledExecutorService scheduledExecutor;
 
   /**
    * Per application lock for synchronizing all internal FirebaseApp state changes.
@@ -107,7 +112,7 @@ public class FirebaseApp {
     this.options = checkNotNull(options);
     this.tokenRefresher = checkNotNull(factory).create(this);
     this.threadManager = options.getThreadManager();
-    this.executor = this.threadManager.getListeningExecutor(this);
+    this.executors = this.threadManager.getFirebaseExecutors(this);
   }
 
   /** Returns a list of all FirebaseApps. */
@@ -323,10 +328,14 @@ public class FirebaseApp {
         service.destroy();
       }
       services.clear();
-      tokenRefresher.cleanup();
+      tokenRefresher.stop();
 
-      // Clean up and terminate the thread pool
-      threadManager.releaseExecutor(this, executor);
+      // Clean up and terminate the thread pools
+      threadManager.releaseFirebaseExecutors(this, executors);
+      if (scheduledExecutor != null) {
+        scheduledExecutor.shutdownNow();
+        scheduledExecutor = null;
+      }
     }
 
     synchronized (appsLock) {
@@ -343,6 +352,19 @@ public class FirebaseApp {
     checkState(!deleted.get(), "FirebaseApp was deleted %s", this);
   }
 
+  private ScheduledExecutorService ensureScheduledExecutorService() {
+    if (scheduledExecutor == null) {
+      synchronized (lock) {
+        checkNotDeleted();
+        if (scheduledExecutor == null) {
+          scheduledExecutor = new RevivingScheduledExecutor(threadManager.getThreadFactory(),
+              "firebase-scheduled-worker", GaeThreadFactory.isAvailable());
+        }
+      }
+    }
+    return scheduledExecutor;
+  }
+
   ThreadFactory getThreadFactory() {
     return threadManager.getThreadFactory();
   }
@@ -350,12 +372,25 @@ public class FirebaseApp {
   // TODO: Return an ApiFuture once Task API is fully removed.
   <T> Task<T> submit(Callable<T> command) {
     checkNotNull(command);
-    return Tasks.call(executor, command);
+    return Tasks.call(executors.getListeningExecutor(), command);
   }
 
   <T> ScheduledFuture<T> schedule(Callable<T> command, long delayMillis) {
     checkNotNull(command);
-    return executor.schedule(command, delayMillis, TimeUnit.MILLISECONDS);
+    try {
+      return ensureScheduledExecutorService().schedule(command, delayMillis, TimeUnit.MILLISECONDS);
+    } catch (Exception e) {
+      // This may fail if the underlying ThreadFactory does not support long-lived threads.
+      throw new UnsupportedOperationException("Scheduled tasks not supported", e);
+    }
+  }
+
+  void startTokenRefresher() {
+    synchronized (lock) {
+      checkNotDeleted();
+      // TODO: Provide an option to disable this altogether.
+      tokenRefresher.start();
+    }
   }
 
   boolean isDefaultApp() {
@@ -390,24 +425,24 @@ public class FirebaseApp {
 
     private final FirebaseApp firebaseApp;
     private final GoogleCredentials credentials;
+    private final AtomicReference<State> state;
+
     private Future<Void> future;
-    private boolean closed;
 
     TokenRefresher(FirebaseApp firebaseApp) {
       this.firebaseApp = checkNotNull(firebaseApp);
       this.credentials = firebaseApp.getOptions().getCredentials();
-      this.credentials.addChangeListener(this);
+      this.state = new AtomicReference<>(State.READY);
     }
 
     @Override
     public final synchronized void onChanged(OAuth2Credentials credentials) throws IOException {
-      if (closed) {
+      if (state.get() != State.STARTED) {
         return;
       }
 
       AccessToken accessToken = credentials.getAccessToken();
-      long refreshDelay = accessToken.getExpirationTime().getTime()
-          - System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(5);
+      long refreshDelay = getRefreshDelay(accessToken);
       if (refreshDelay > 0) {
         scheduleRefresh(refreshDelay);
       } else {
@@ -416,26 +451,6 @@ public class FirebaseApp {
       }
     }
 
-    /**
-     * Schedule a forced token refresh to be executed after a specified duration.
-     *
-     * @param delayMillis Duration in milliseconds, after which the token should be forcibly
-     *     refreshed.
-     */
-    private void scheduleRefresh(final long delayMillis) {
-      cancelPrevious();
-      scheduleNext(
-          new Callable<Void>() {
-            @Override
-            public Void call() throws Exception {
-              logger.debug("Refreshing OAuth2 credential");
-              credentials.refresh();
-              return null;
-            }
-          },
-          delayMillis);
-    }
-
     protected void cancelPrevious() {
       if (future != null) {
         future.cancel(true);
@@ -446,14 +461,73 @@ public class FirebaseApp {
       logger.debug("Scheduling next token refresh in {} milliseconds", delayMillis);
       try {
         future = firebaseApp.schedule(task, delayMillis);
-      } catch (UnsupportedOperationException ignored) {
+      } catch (UnsupportedOperationException e) {
         // Cannot support task scheduling in the current runtime.
+        logger.debug("Failed to schedule token refresh event", e);
+      }
+    }
+
+    /**
+     * Starts the TokenRefresher if not already started. Starts listening to credentials changed
+     * events, and schedules refresh events every time the OAuth2 token changes. If no active
+     * token is present, or if the available token is set to expire soon, this will also schedule
+     * a refresh event to be executed immediately.
+     *
+     * <p>This operation is idempotent. Calling it multiple times, or calling it after the
+     * refresher has been stopped has no effect.
+     */
+    final synchronized void start() {
+      // Allow starting only from the ready state.
+      if (!state.compareAndSet(State.READY, State.STARTED)) {
+        return;
+      }
+
+      logger.debug("Starting the proactive token refresher");
+      credentials.addChangeListener(this);
+      AccessToken accessToken = credentials.getAccessToken();
+      long refreshDelay;
+      if (accessToken != null) {
+        // If the token is about to expire (i.e. expires in less than 5 minutes), schedule a
+        // refresh event with 0 delay. Otherwise schedule a refresh event at the regular token
+        // expiry time, minus 5 minutes.
+        refreshDelay = Math.max(getRefreshDelay(accessToken), 0L);
+      } else {
+        // If there is no token fetched so far, fetch one immediately.
+        refreshDelay = 0L;
+      }
+      scheduleRefresh(refreshDelay);
+    }
+
+    final synchronized void stop() {
+      // Allow stopping from any state.
+      State previous = state.getAndSet(State.STOPPED);
+      if (previous == State.STARTED) {
+        cancelPrevious();
+        logger.debug("Stopped the proactive token refresher");
       }
     }
 
-    protected synchronized void cleanup() {
+    /**
+     * Schedule a forced token refresh to be executed after a specified duration.
+     *
+     * @param delayMillis Duration in milliseconds, after which the token should be forcibly
+     *     refreshed.
+     */
+    private void scheduleRefresh(final long delayMillis) {
       cancelPrevious();
-      closed = true;
+      scheduleNext(new Callable<Void>() {
+        @Override
+        public Void call() throws Exception {
+          logger.debug("Refreshing OAuth2 credential");
+          credentials.refresh();
+          return null;
+        }
+      }, delayMillis);
+    }
+
+    private long getRefreshDelay(AccessToken accessToken) {
+      return accessToken.getExpirationTime().getTime() - System.currentTimeMillis()
+          - TimeUnit.MINUTES.toMillis(5);
     }
 
     static class Factory {
@@ -461,5 +535,11 @@ public class FirebaseApp {
         return new TokenRefresher(app);
       }
     }
+
+    enum State {
+      READY,
+      STARTED,
+      STOPPED
+    }
   }
 }
diff --git a/src/main/java/com/google/firebase/FirebaseOptions.java b/src/main/java/com/google/firebase/FirebaseOptions.java
index a485516..d77bec4 100644
--- a/src/main/java/com/google/firebase/FirebaseOptions.java
+++ b/src/main/java/com/google/firebase/FirebaseOptions.java
@@ -28,7 +28,7 @@ import com.google.firebase.auth.FirebaseCredential;
 import com.google.firebase.auth.FirebaseCredentials;
 import com.google.firebase.auth.internal.BaseCredential;
 import com.google.firebase.auth.internal.FirebaseCredentialsAdapter;
-import com.google.firebase.internal.FirebaseExecutors;
+import com.google.firebase.internal.FirebaseThreadManagers;
 import com.google.firebase.internal.NonNull;
 import com.google.firebase.internal.Nullable;
 
@@ -139,7 +139,7 @@ public final class FirebaseOptions {
     private String projectId;
     private HttpTransport httpTransport = Utils.getDefaultTransport();
     private JsonFactory jsonFactory = Utils.getDefaultJsonFactory();
-    private ThreadManager threadManager = FirebaseExecutors.DEFAULT_THREAD_MANAGER;
+    private ThreadManager threadManager = FirebaseThreadManagers.DEFAULT_THREAD_MANAGER;
 
     /** Constructs an empty builder. */
     public Builder() {}
diff --git a/src/main/java/com/google/firebase/ImplFirebaseTrampolines.java b/src/main/java/com/google/firebase/ImplFirebaseTrampolines.java
index 65a8a10..7ffac6d 100644
--- a/src/main/java/com/google/firebase/ImplFirebaseTrampolines.java
+++ b/src/main/java/com/google/firebase/ImplFirebaseTrampolines.java
@@ -21,10 +21,7 @@ import com.google.firebase.internal.FirebaseService;
 import com.google.firebase.internal.NonNull;
 
 import com.google.firebase.tasks.Task;
-import com.google.firebase.tasks.Tasks;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 
 /**
@@ -76,4 +73,8 @@ public final class ImplFirebaseTrampolines {
   public static <T> Task<T> submitCallable(@NonNull FirebaseApp app, @NonNull Callable<T> command) {
     return app.submit(command);
   }
+
+  public static void startTokenRefresher(@NonNull FirebaseApp app) {
+    app.startTokenRefresher();
+  }
 }
diff --git a/src/main/java/com/google/firebase/ThreadManager.java b/src/main/java/com/google/firebase/ThreadManager.java
index 0c65ad1..343bbd7 100644
--- a/src/main/java/com/google/firebase/ThreadManager.java
+++ b/src/main/java/com/google/firebase/ThreadManager.java
@@ -18,11 +18,11 @@ package com.google.firebase;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
-import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.firebase.internal.NonNull;
 
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadFactory;
 
 /**
@@ -36,42 +36,72 @@ import java.util.concurrent.ThreadFactory;
 public abstract class ThreadManager {
 
   @NonNull
-  final ListeningScheduledExecutorService getListeningExecutor(@NonNull FirebaseApp app) {
-    ScheduledExecutorService executor = getExecutor(app);
-    checkNotNull(executor, "ScheduledExecutorService must not be null");
-    return MoreExecutors.listeningDecorator(executor);
+  final FirebaseExecutors getFirebaseExecutors(@NonNull FirebaseApp app) {
+    return new FirebaseExecutors(getExecutor(app));
+  }
+
+  final void releaseFirebaseExecutors(
+      @NonNull FirebaseApp app, @NonNull FirebaseExecutors executor) {
+    releaseExecutor(app, executor.userExecutor);
   }
 
   /**
    * Returns the main thread pool for an app. Implementations may return the same instance of
-   * <code>ScheduledExecutorService</code> for multiple apps. The returned thread pool is used by
-   * all components of an app except for the Realtime Database. Database has far stricter and
-   * complicated threading requirements, and thus initializes its own threads using the
-   * factory returned by {@link ThreadManager#getThreadFactory()}.
+   * <code>ExecutorService</code> for multiple apps. The returned thread pool is used for
+   * short-lived tasks by all components of an app.
+   *
+   * <p>For long-lived tasks (such as the ones started by the Realtime Database client), the SDK
+   * creates dedicated executors using the <code>ThreadFactory</code> returned by the
+   * {@link #getThreadFactory()} method.
    *
    * @param app A {@link FirebaseApp} instance.
-   * @return A non-null {@link ScheduledExecutorService} instance.
+   * @return A non-null {@link ExecutorService} instance.
    */
   @NonNull
-  protected abstract ScheduledExecutorService getExecutor(@NonNull FirebaseApp app);
+  protected abstract ExecutorService getExecutor(@NonNull FirebaseApp app);
 
   /**
-   * Cleans up the thread pool associated with an app. This method is invoked when an
-   * app is deleted.
+   * Cleans up the thread pool associated with an app. This method is invoked when an app
+   * is deleted. This is guaranteed to be called with the <code>ExecutorService</code> previously
+   * returned by {@link #getExecutor(FirebaseApp)} for the corresponding app.
    *
    * @param app A {@link FirebaseApp} instance.
    */
   protected abstract void releaseExecutor(
-      @NonNull FirebaseApp app, @NonNull ScheduledExecutorService executor);
+      @NonNull FirebaseApp app, @NonNull ExecutorService executor);
 
   /**
-   * Returns the <code>ThreadFactory</code> to be used for creating any additional threads
-   * required by the SDK. This is used mainly to create the long-lived worker threads for
-   * the Realtime Database client.
+   * Returns the <code>ThreadFactory</code> to be used for creating long-lived threads. This is
+   * used mainly to create the long-lived worker threads for the Realtime Database client, and
+   * other scheduled (periodic) tasks started by the SDK. The SDK guarantees
+   * clean termination of all threads started via this <code>ThreadFactory</code>, upon
+   * calling {@link FirebaseApp#delete()}.
+   *
+   * <p>If long-lived threads cannot be supported in the current runtime, this method may
+   * throw a RuntimeException.
    *
    * @return A non-null <code>ThreadFactory</code>.
    */
   @NonNull
   protected abstract ThreadFactory getThreadFactory();
 
+  /**
+   * Wraps an ExecutorService in a ListeningExecutorService while keeping a reference to the
+   * original ExecutorService. This reference is used when it's time to release/cleanup the
+   * original ExecutorService.
+   */
+  static final class FirebaseExecutors {
+    private final ExecutorService userExecutor;
+    private final ListeningExecutorService listeningExecutor;
+
+    private FirebaseExecutors(ExecutorService userExecutor) {
+      this.userExecutor = checkNotNull(userExecutor, "ExecutorService must not be null");
+      this.listeningExecutor = MoreExecutors.listeningDecorator(userExecutor);
+    }
+
+    ListeningExecutorService getListeningExecutor() {
+      return listeningExecutor;
+    }
+  }
+
 }
diff --git a/src/main/java/com/google/firebase/database/core/JvmAuthTokenProvider.java b/src/main/java/com/google/firebase/database/core/JvmAuthTokenProvider.java
index 7301b07..541a4e4 100644
--- a/src/main/java/com/google/firebase/database/core/JvmAuthTokenProvider.java
+++ b/src/main/java/com/google/firebase/database/core/JvmAuthTokenProvider.java
@@ -28,18 +28,25 @@ import com.google.firebase.database.util.GAuthToken;
 
 import java.io.IOException;
 import java.util.Map;
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Executor;
 
 public class JvmAuthTokenProvider implements AuthTokenProvider {
 
   private final GoogleCredentials credentials;
   private final Map<String, Object> authVariable;
-  private final ScheduledExecutorService executorService;
+  private final Executor executor;
 
-  JvmAuthTokenProvider(FirebaseApp firebaseApp, ScheduledExecutorService executorService) {
+  JvmAuthTokenProvider(FirebaseApp firebaseApp, Executor executor) {
+    this(firebaseApp, executor, true);
+  }
+
+  JvmAuthTokenProvider(FirebaseApp firebaseApp, Executor executor, boolean autoRefresh) {
     this.credentials = ImplFirebaseTrampolines.getCredentials(firebaseApp);
     this.authVariable = firebaseApp.getOptions().getDatabaseAuthVariableOverride();
-    this.executorService = executorService;
+    this.executor = executor;
+    if (autoRefresh) {
+      ImplFirebaseTrampolines.startTokenRefresher(firebaseApp);
+    }
   }
 
   @Override
@@ -57,7 +64,7 @@ public class JvmAuthTokenProvider implements AuthTokenProvider {
 
       AccessToken accessToken = credentials.getAccessToken();
       listener.onSuccess(wrapOAuthToken(accessToken, authVariable));
-    } catch (IOException e) {
+    } catch (Exception e) {
       listener.onError(e.toString());
     }
   }
@@ -65,7 +72,7 @@ public class JvmAuthTokenProvider implements AuthTokenProvider {
   @Override
   public void addTokenChangeListener(TokenChangeListener listener) {
     CredentialsChangedListener listenerWrapper = new TokenChangeListenerWrapper(
-        listener, executorService, authVariable);
+        listener, executor, authVariable);
     credentials.addChangeListener(listenerWrapper);
   }
 
@@ -87,15 +94,15 @@ public class JvmAuthTokenProvider implements AuthTokenProvider {
   private static class TokenChangeListenerWrapper implements CredentialsChangedListener {
 
     private final TokenChangeListener listener;
-    private final ScheduledExecutorService executorService;
+    private final Executor executor;
     private final Map<String, Object> authVariable;
 
     TokenChangeListenerWrapper(
         TokenChangeListener listener,
-        ScheduledExecutorService executorService,
+        Executor executor,
         Map<String, Object> authVariable) {
       this.listener = checkNotNull(listener, "Listener must not be null");
-      this.executorService = checkNotNull(executorService, "ExecutorService must not be null");
+      this.executor = checkNotNull(executor, "Executor must not be null");
       this.authVariable = authVariable;
     }
 
@@ -107,7 +114,7 @@ public class JvmAuthTokenProvider implements AuthTokenProvider {
 
       // Notify the TokenChangeListener on database's thread pool to make sure that
       // all database work happens on database worker threads.
-      executorService.execute(
+      executor.execute(
           new Runnable() {
             @Override
             public void run() {
diff --git a/src/main/java/com/google/firebase/internal/FirebaseExecutors.java b/src/main/java/com/google/firebase/internal/FirebaseThreadManagers.java
similarity index 57%
rename from src/main/java/com/google/firebase/internal/FirebaseExecutors.java
rename to src/main/java/com/google/firebase/internal/FirebaseThreadManagers.java
index 1044a38..0e503ba 100644
--- a/src/main/java/com/google/firebase/internal/FirebaseExecutors.java
+++ b/src/main/java/com/google/firebase/internal/FirebaseThreadManagers.java
@@ -24,17 +24,17 @@ import com.google.firebase.ThreadManager;
 
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /** Default ThreadManager implementations used by the Admin SDK. */
-public class FirebaseExecutors {
+public class FirebaseThreadManagers {
 
-  private static final Logger logger = LoggerFactory.getLogger(FirebaseExecutors.class);
+  private static final Logger logger = LoggerFactory.getLogger(FirebaseThreadManagers.class);
 
   public static final ThreadManager DEFAULT_THREAD_MANAGER;
 
@@ -55,11 +55,10 @@ public class FirebaseExecutors {
 
     private final Object lock = new Object();
     private final Set<String> apps = new HashSet<>();
-
-    private ScheduledExecutorService executorService;
+    private ExecutorService executorService;
 
     @Override
-    protected ScheduledExecutorService getExecutor(FirebaseApp app) {
+    protected ExecutorService getExecutor(FirebaseApp app) {
       synchronized (lock) {
         if (executorService == null) {
           executorService = doInit();
@@ -70,7 +69,7 @@ public class FirebaseExecutors {
     }
 
     @Override
-    protected void releaseExecutor(FirebaseApp app, ScheduledExecutorService executor) {
+    protected void releaseExecutor(FirebaseApp app, ExecutorService executor) {
       synchronized (lock) {
         if (apps.remove(app.getName()) && apps.isEmpty()) {
           doCleanup(executorService);
@@ -82,31 +81,29 @@ public class FirebaseExecutors {
     /**
      * Initializes the executor service. Called when the first application is initialized.
      */
-    protected abstract ScheduledExecutorService doInit();
+    protected abstract ExecutorService doInit();
 
     /**
      * Cleans up the executor service. Called when the last application is deleted.
      */
-    protected abstract void doCleanup(ScheduledExecutorService executorService);
+    protected abstract void doCleanup(ExecutorService executorService);
   }
 
   private static class DefaultThreadManager extends GlobalThreadManager {
 
     @Override
-    protected ScheduledExecutorService doInit() {
-      int cores = Runtime.getRuntime().availableProcessors();
+    protected ExecutorService doInit() {
       // Create threads as daemons to ensure JVM exit when all foreground jobs are complete.
       ThreadFactory threadFactory = new ThreadFactoryBuilder()
           .setNameFormat("firebase-default-%d")
           .setDaemon(true)
           .setThreadFactory(getThreadFactory())
           .build();
-      logger.debug("Initializing default executor with {} max threads", cores);
-      return Executors.newScheduledThreadPool(cores, threadFactory);
+      return Executors.newCachedThreadPool(threadFactory);
     }
 
     @Override
-    protected void doCleanup(ScheduledExecutorService executorService) {
+    protected void doCleanup(ExecutorService executorService) {
       logger.debug("Shutting down default executor");
       executorService.shutdownNow();
     }
@@ -117,25 +114,50 @@ public class FirebaseExecutors {
     }
   }
 
+  /**
+   * The ThreadManager implementation that will be used by default in the Google App Engine
+   * environment.
+   *
+   * <p>Auto-scaling: Creates an ExecutorService backed by the request-scoped ThreadFactory. This
+   * can be used for any short-lived task, such as the ones submitted by components like
+   * FirebaseAuth. {@link #getThreadFactory()} throws an exception, since long-lived threads
+   * cannot be supported. Therefore task scheduling and RTDB will not work.
+   *
+   * <p>Manual-scaling: Creates a single-threaded ExecutorService backed by the background
+   * ThreadFactory. Keeps the threads alive indefinitely by periodically restarting them (see
+   * {@link RevivingScheduledExecutor}). Threads will be terminated only when the method
+   * {@link #releaseExecutor(FirebaseApp, ExecutorService)} is invoked. The
+   * {@link #getThreadFactory()} also returns the background ThreadFactory enabling other
+   * components in the SDK to start long-lived threads when necessary. Therefore task scheduling
+   * and RTDB can be supported as if running on the regular JVM.
+   *
+   * <p>Basic-scaling: Behavior is similar to manual-scaling. Since the threads are kept alive
+   * indefinitely, prevents the GAE idle instance shutdown. Developers are advised to use
+   * a custom ThreadManager implementation if idle instance shutdown should be supported. In
+   * general, a ThreadManager implementation that uses the request-scoped ThreadFactory, or the
+   * background ThreadFactory with specific keep-alive times can easily facilitate GAE idle
+   * instance shutdown. Note that this often comes at the cost of losing scheduled tasks and RTDB
+   * support. Therefore, for these features, manual-scaling is the recommended GAE deployment mode
+   * regardless of the ThreadManager implementation used.
+   */
   private static class GaeThreadManager extends GlobalThreadManager {
 
     @Override
-    protected ScheduledExecutorService doInit() {
-      return new GaeScheduledExecutorService("gae-firebase-default");
+    protected ExecutorService doInit() {
+      return new GaeExecutorService("gae-firebase-default");
     }
 
     @Override
-    protected void doCleanup(ScheduledExecutorService executorService) {
+    protected void doCleanup(ExecutorService executorService) {
       executorService.shutdownNow();
     }
 
     @Override
-
     protected ThreadFactory getThreadFactory() {
       GaeThreadFactory threadFactory = GaeThreadFactory.getInstance();
       checkState(threadFactory.isUsingBackgroundThreads(),
           "Failed to initialize a GAE background thread factory. Background thread support "
-              + "is required to access the Realtime database from App Engine environment.");
+              + "is required to create long-lived threads.");
       return threadFactory;
     }
   }
diff --git a/src/main/java/com/google/firebase/internal/GaeScheduledExecutorService.java b/src/main/java/com/google/firebase/internal/GaeExecutorService.java
similarity index 65%
rename from src/main/java/com/google/firebase/internal/GaeScheduledExecutorService.java
rename to src/main/java/com/google/firebase/internal/GaeExecutorService.java
index 3be11e9..6ee3b16 100644
--- a/src/main/java/com/google/firebase/internal/GaeScheduledExecutorService.java
+++ b/src/main/java/com/google/firebase/internal/GaeExecutorService.java
@@ -28,8 +28,6 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -37,26 +35,24 @@ import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
- * A ScheduledExecutorService instance that can operate in the Google App Engine environment. The
- * scheduling operations (i.e. operations specific to the ScheduledExecutorService interface) can
- * only be used when background thread support is enabled. These operations will throw
- * UnsupportedOperationException when invoked in an auto-scaled instance without background thread
- * support. Operations inherited from the ExecutorService and Executor interfaces will work
- * regardless of the background threads support. This implementation is also lazy loaded to prevent
- * unnecessary RPC calls to the GAE backend.
+ * An ExecutorService instance that can operate in the Google App Engine environment. When
+ * available, uses background thread support to initialize an ExecutorService with long-lived
+ * threads. Otherwise, creates an ExecutorService that spawns short-lived threads as tasks
+ * are submitted. The actual ExecutorService implementation is lazy-loaded to prevent making
+ * unnecessary RPC calls to the GAE's native ThreadFactory mechanism.
  */
-class GaeScheduledExecutorService implements ScheduledExecutorService {
+class GaeExecutorService implements ExecutorService {
 
   private final AtomicReference<ExecutorWrapper> executor = new AtomicReference<>();
   private final String threadName;
   private boolean shutdown;
 
-  GaeScheduledExecutorService(String threadName) {
+  GaeExecutorService(String threadName) {
     checkArgument(!Strings.isNullOrEmpty(threadName));
     this.threadName = threadName;
   }
 
-  private ExecutorWrapper ensureExecutorWrapper() {
+  private ExecutorService ensureExecutorService() {
     ExecutorWrapper wrapper = executor.get();
     if (wrapper == null) {
       synchronized (executor) {
@@ -68,47 +64,7 @@ class GaeScheduledExecutorService implements ScheduledExecutorService {
         }
       }
     }
-    return wrapper;
-  }
-
-  private ExecutorService ensureExecutorService() {
-    return ensureExecutorWrapper().getExecutorService();
-  }
-
-  private ScheduledExecutorService ensureScheduledExecutorService() {
-    ScheduledExecutorService scheduledExecutorService =
-        ensureExecutorWrapper().getScheduledExecutorService();
-    if (scheduledExecutorService != null) {
-      return scheduledExecutorService;
-    } else {
-      throw new UnsupportedOperationException(
-          "ScheduledExecutorService not available. A manually-scaled instance is "
-          + "required when running the Firebase Admin SDK on GAE.");
-    }
-  }
-
-  @Override
-  public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
-    return ensureScheduledExecutorService().schedule(command, delay, unit);
-  }
-
-  @Override
-  public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
-    return ensureScheduledExecutorService().schedule(callable, delay, unit);
-  }
-
-  @Override
-  public ScheduledFuture<?> scheduleAtFixedRate(
-      Runnable command, long initialDelay, long period, TimeUnit unit) {
-    return ensureScheduledExecutorService()
-        .scheduleAtFixedRate(command, initialDelay, period, unit);
-  }
-
-  @Override
-  public ScheduledFuture<?> scheduleWithFixedDelay(
-      Runnable command, long initialDelay, long delay, TimeUnit unit) {
-    return ensureScheduledExecutorService()
-        .scheduleWithFixedDelay(command, initialDelay, delay, unit);
+    return wrapper.getExecutorService();
   }
 
   @Override
@@ -200,18 +156,15 @@ class GaeScheduledExecutorService implements ScheduledExecutorService {
   private static class ExecutorWrapper {
 
     private final ExecutorService executorService;
-    private final ScheduledExecutorService scheduledExecutorService;
 
     ExecutorWrapper(String threadName) {
       GaeThreadFactory threadFactory = GaeThreadFactory.getInstance();
       if (threadFactory.isUsingBackgroundThreads()) {
         // Create a thread pool with long-lived threads if background thread support is available.
-        scheduledExecutorService = new RevivingScheduledExecutor(threadFactory, threadName, true);
-        executorService = scheduledExecutorService;
+        executorService = new RevivingScheduledExecutor(threadFactory, threadName, true);
       } else {
         // Create an executor that creates a new thread for each submitted task, when background
         // thread support is not available.
-        scheduledExecutorService = null;
         executorService =
             new ThreadPoolExecutor(
                 0,
@@ -226,9 +179,5 @@ class GaeScheduledExecutorService implements ScheduledExecutorService {
     ExecutorService getExecutorService() {
       return executorService;
     }
-
-    ScheduledExecutorService getScheduledExecutorService() {
-      return scheduledExecutorService;
-    }
   }
 }
diff --git a/src/main/java/com/google/firebase/internal/GaeThreadFactory.java b/src/main/java/com/google/firebase/internal/GaeThreadFactory.java
index f7e5b3f..01d0dfa 100644
--- a/src/main/java/com/google/firebase/internal/GaeThreadFactory.java
+++ b/src/main/java/com/google/firebase/internal/GaeThreadFactory.java
@@ -19,7 +19,7 @@ package com.google.firebase.internal;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.lang.reflect.InvocationTargetException;
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
@@ -35,8 +35,8 @@ public class GaeThreadFactory implements ThreadFactory {
 
   private static final Logger logger = LoggerFactory.getLogger(GaeThreadFactory.class);
 
-  public static final ScheduledExecutorService DEFAULT_EXECUTOR =
-      new GaeScheduledExecutorService("LegacyFirebaseDefault");
+  public static final ExecutorService DEFAULT_EXECUTOR =
+      new GaeExecutorService("LegacyFirebaseDefault");
   private static final String GAE_THREAD_MANAGER_CLASS = "com.google.appengine.api.ThreadManager";
   private static final GaeThreadFactory instance = new GaeThreadFactory();
   private final AtomicReference<ThreadFactoryWrapper> threadFactory = new AtomicReference<>(null);
@@ -111,7 +111,7 @@ public class GaeThreadFactory implements ThreadFactory {
     // thread factories here and discard one once we detect that we are running in an
     // automatically scaled instance.
     //
-    // Note: It's fine if multiple thread access this block at the same time.
+    // Note: It's fine if multiple threads access this block at the same time.
     try {
       try {
         threadFactory = createBackgroundFactory();


Test patch:
diff --git a/src/test/java/com/google/firebase/FirebaseAppTest.java b/src/test/java/com/google/firebase/FirebaseAppTest.java
index 07735f6..732db65 100644
--- a/src/test/java/com/google/firebase/FirebaseAppTest.java
+++ b/src/test/java/com/google/firebase/FirebaseAppTest.java
@@ -332,7 +332,11 @@ public class FirebaseAppTest {
     CredentialsChangedListener listener = mock(CredentialsChangedListener.class);
     ImplFirebaseTrampolines.getCredentials(firebaseApp).addChangeListener(listener);
 
-    TestOnlyImplFirebaseTrampolines.getToken(firebaseApp, true);
+    firebaseApp.startTokenRefresher();
+
+    // Since there was no token to begin with, the refresher should refresh the credential
+    // immediately.
+    tokenRefresher.simulateDelay(0);
     verify(listener, times(1)).onChanged(Mockito.any(OAuth2Credentials.class));
 
     tokenRefresher.simulateDelay(55);
@@ -345,6 +349,76 @@ public class FirebaseAppTest {
     verify(listener, times(3)).onChanged(Mockito.any(OAuth2Credentials.class));
   }
 
+  @Test
+  public void testProactiveTokenRefreshWithInitialToken() throws Exception {
+    MockTokenRefresherFactory factory = new MockTokenRefresherFactory();
+    FirebaseApp firebaseApp = FirebaseApp.initializeApp(getMockCredentialOptions(), "myApp",
+        factory);
+    MockTokenRefresher tokenRefresher = factory.instance;
+    Assert.assertNotNull(tokenRefresher);
+
+    // Get the initial token
+    TestOnlyImplFirebaseTrampolines.getToken(firebaseApp, true);
+
+    CredentialsChangedListener listener = mock(CredentialsChangedListener.class);
+    ImplFirebaseTrampolines.getCredentials(firebaseApp).addChangeListener(listener);
+
+    firebaseApp.startTokenRefresher();
+
+    // Since there is already a valid token, which won't expire for another hour, the refresher
+    // should not refresh the credential at this point in time.
+    tokenRefresher.simulateDelay(0);
+    verify(listener, never()).onChanged(Mockito.any(OAuth2Credentials.class));
+
+    tokenRefresher.simulateDelay(55);
+    verify(listener, times(1)).onChanged(Mockito.any(OAuth2Credentials.class));
+
+    tokenRefresher.simulateDelay(20);
+    verify(listener, times(1)).onChanged(Mockito.any(OAuth2Credentials.class));
+
+    tokenRefresher.simulateDelay(35);
+    verify(listener, times(2)).onChanged(Mockito.any(OAuth2Credentials.class));
+  }
+
+  @Test
+  public void testTokenRefresherStateMachine() {
+    FirebaseApp firebaseApp = FirebaseApp.initializeApp(getMockCredentialOptions(), "myApp");
+    CountingTokenRefresher refresher = new CountingTokenRefresher(firebaseApp);
+    assertEquals(0, refresher.scheduleCalls);
+
+    // Both schedule and cancel should be called.
+    // (we call cancel before each new scheduling attempt)
+    refresher.start();
+    assertEquals(1, refresher.scheduleCalls);
+    assertEquals(1, refresher.cancelCalls);
+
+    // Shouldn't change state
+    refresher.start();
+    assertEquals(1, refresher.scheduleCalls);
+    assertEquals(1, refresher.cancelCalls);
+
+    refresher.stop();
+    assertEquals(1, refresher.scheduleCalls);
+    assertEquals(2, refresher.cancelCalls);
+
+    // Shouldn't change state
+    refresher.start();
+    assertEquals(1, refresher.scheduleCalls);
+    assertEquals(2, refresher.cancelCalls);
+
+    refresher.stop();
+    assertEquals(1, refresher.scheduleCalls);
+    assertEquals(2, refresher.cancelCalls);
+
+    // Test for the case where the refresher is stopped without ever starting.
+    refresher = new CountingTokenRefresher(firebaseApp);
+    // stop() is allowed here, but since we didn't start(), no measurable state change
+    // should take place.
+    refresher.stop();
+    assertEquals(0, refresher.scheduleCalls);
+    assertEquals(0, refresher.cancelCalls);
+  }
+
   @Test(expected = IllegalArgumentException.class)
   public void testFirebaseExceptionNullDetail() {
     new FirebaseException(null);
@@ -392,6 +466,26 @@ public class FirebaseAppTest {
     }
   }
 
+  private static class CountingTokenRefresher extends TokenRefresher {
+
+    private int cancelCalls = 0;
+    private int scheduleCalls = 0;
+
+    CountingTokenRefresher(FirebaseApp firebaseApp) {
+      super(firebaseApp);
+    }
+
+    @Override
+    protected void cancelPrevious() {
+      cancelCalls++;
+    }
+
+    @Override
+    protected void scheduleNext(Callable<Void> task, long delayMillis) {
+      scheduleCalls++;
+    }
+  }
+
   private static class MockTokenRefresherFactory extends TokenRefresher.Factory {
 
     MockTokenRefresher instance;
@@ -403,7 +497,6 @@ public class FirebaseAppTest {
     }
   }
 
-
   private static class MockGoogleCredentials extends GoogleCredentials {
 
     @Override
diff --git a/src/test/java/com/google/firebase/FirebaseOptionsTest.java b/src/test/java/com/google/firebase/FirebaseOptionsTest.java
index f043336..1d39fde 100644
--- a/src/test/java/com/google/firebase/FirebaseOptionsTest.java
+++ b/src/test/java/com/google/firebase/FirebaseOptionsTest.java
@@ -36,7 +36,7 @@ import com.google.firebase.tasks.Task;
 import com.google.firebase.testing.ServiceAccount;
 import com.google.firebase.testing.TestUtils;
 import java.io.IOException;
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadFactory;
 
 import org.junit.Test;
@@ -60,12 +60,12 @@ public class FirebaseOptionsTest {
 
   private static final ThreadManager MOCK_THREAD_MANAGER = new ThreadManager() {
     @Override
-    protected ScheduledExecutorService getExecutor(FirebaseApp app) {
+    protected ExecutorService getExecutor(FirebaseApp app) {
       return null;
     }
 
     @Override
-    protected void releaseExecutor(FirebaseApp app, ScheduledExecutorService executor) {
+    protected void releaseExecutor(FirebaseApp app, ExecutorService executor) {
     }
 
     @Override
diff --git a/src/test/java/com/google/firebase/ThreadManagerTest.java b/src/test/java/com/google/firebase/ThreadManagerTest.java
index e89896e..d8689a4 100644
--- a/src/test/java/com/google/firebase/ThreadManagerTest.java
+++ b/src/test/java/com/google/firebase/ThreadManagerTest.java
@@ -17,6 +17,7 @@
 package com.google.firebase;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 
 import com.google.firebase.auth.FirebaseAuth;
 import com.google.firebase.auth.MockGoogleCredentials;
@@ -25,22 +26,31 @@ import com.google.firebase.internal.Nullable;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
 
 public class ThreadManagerTest {
 
+  private ExecutorService executor;
+
+  @Before
+  public void setUp() {
+    executor = Executors.newSingleThreadExecutor();
+  }
+
   @After
   public void tearDown() {
     TestOnlyImplFirebaseTrampolines.clearInstancesForTest();
+    executor.shutdownNow();
   }
 
   @Test
   public void testBareBonesAppLifecycle() {
-    MockThreadManager threadManager = new MockThreadManager();
+    MockThreadManager threadManager = new MockThreadManager(executor);
 
     // Initializing an app should initialize the executor.
     FirebaseApp app = FirebaseApp.initializeApp(buildOptions(threadManager));
@@ -59,7 +69,7 @@ public class ThreadManagerTest {
 
   @Test
   public void testBareBonesAppLifecycleWithMultipleApps() {
-    MockThreadManager threadManager = new MockThreadManager();
+    MockThreadManager threadManager = new MockThreadManager(executor);
 
     // Initializing an app should initialize the executor.
     final FirebaseApp app1 = FirebaseApp.initializeApp(buildOptions(threadManager));
@@ -90,14 +100,14 @@ public class ThreadManagerTest {
 
   @Test
   public void testAppLifecycleWithExecutor() {
-    MockThreadManager threadManager = new MockThreadManager();
+    MockThreadManager threadManager = new MockThreadManager(executor);
 
     // Initializing an app should initialize the executor.
     FirebaseApp app = FirebaseApp.initializeApp(buildOptions(threadManager));
     assertEquals(1, threadManager.events.size());
     Event event = threadManager.events.get(0);
     assertEquals(Event.TYPE_GET_EXECUTOR, event.type);
-    assertEquals(app, event.app);
+    assertSame(app, event.app);
 
     // Should not re-initialize
     TestOnlyImplFirebaseTrampolines.forceThreadManagerInit(app);
@@ -108,12 +118,13 @@ public class ThreadManagerTest {
     assertEquals(2, threadManager.events.size());
     event = threadManager.events.get(1);
     assertEquals(Event.TYPE_RELEASE_EXECUTOR, event.type);
-    assertEquals(app, event.app);
+    assertSame(app, event.app);
+    assertSame(executor, event.executor);
   }
 
   @Test
   public void testAppLifecycleWithTokenRefresh() {
-    MockThreadManager threadManager = new MockThreadManager();
+    MockThreadManager threadManager = new MockThreadManager(executor);
 
     // Initializing an app should initialize the executor.
     FirebaseApp app = FirebaseApp.initializeApp(buildOptions(threadManager));
@@ -122,26 +133,27 @@ public class ThreadManagerTest {
     assertEquals(Event.TYPE_GET_EXECUTOR, event.type);
     assertEquals(app, event.app);
 
-    // Refreshing the token this way kicks off the proactive token refresh task, which initializes
-    // the executor to schedule the next refresh task.
-    TestOnlyImplFirebaseTrampolines.getToken(app, true);
-    assertEquals(1, threadManager.events.size());
+    // Starting the token refresher should start a long-lived thread using the ThreadFactory.
+    app.startTokenRefresher();
+    assertEquals(2, threadManager.events.size());
+    event = threadManager.events.get(1);
+    assertEquals(Event.TYPE_GET_THREAD_FACTORY, event.type);
 
-    // Should not re-initialize
     TestOnlyImplFirebaseTrampolines.forceThreadManagerInit(app);
-    assertEquals(1, threadManager.events.size());
+    assertEquals(2, threadManager.events.size());
 
     // Deleting app should tear down threading resources.
     app.delete();
-    assertEquals(2, threadManager.events.size());
-    event = threadManager.events.get(1);
+    assertEquals(3, threadManager.events.size());
+    event = threadManager.events.get(2);
     assertEquals(Event.TYPE_RELEASE_EXECUTOR, event.type);
-    assertEquals(app, event.app);
+    assertSame(app, event.app);
+    assertSame(executor, event.executor);
   }
 
   @Test
   public void testAppLifecycleWithServiceCall() {
-    MockThreadManager threadManager = new MockThreadManager();
+    MockThreadManager threadManager = new MockThreadManager(executor);
 
     // Initializing an app should initialize the executor.
     FirebaseApp app = FirebaseApp.initializeApp(buildOptions(threadManager));
@@ -161,19 +173,23 @@ public class ThreadManagerTest {
       // ignored
     }
     assertEquals(1, threadManager.events.size());
+    event = threadManager.events.get(0);
+    assertEquals(Event.TYPE_GET_EXECUTOR, event.type);
+    assertSame(app, event.app);
 
     // Deleting app should tear down threading resources.
     app.delete();
     assertEquals(2, threadManager.events.size());
     event = threadManager.events.get(1);
     assertEquals(Event.TYPE_RELEASE_EXECUTOR, event.type);
-    assertEquals(app, event.app);
+    assertSame(app, event.app);
+    assertSame(executor, event.executor);
   }
 
   @Test
   public void testAppLifecycleWithMultipleServiceCalls()
       throws ExecutionException, InterruptedException {
-    MockThreadManager threadManager = new MockThreadManager();
+    MockThreadManager threadManager = new MockThreadManager(executor);
 
     // Initializing an app should initialize the executor.
     FirebaseApp app = FirebaseApp.initializeApp(buildOptions(threadManager));
@@ -195,7 +211,7 @@ public class ThreadManagerTest {
     assertEquals(1, threadManager.events.size());
     event = threadManager.events.get(0);
     assertEquals(Event.TYPE_GET_EXECUTOR, event.type);
-    assertEquals(app, event.app);
+    assertSame(app, event.app);
 
     // Calling a method again should not re-initialize threading resources.
     try {
@@ -214,7 +230,8 @@ public class ThreadManagerTest {
     assertEquals(2, threadManager.events.size());
     event = threadManager.events.get(1);
     assertEquals(Event.TYPE_RELEASE_EXECUTOR, event.type);
-    assertEquals(app, event.app);
+    assertSame(app, event.app);
+    assertSame(executor, event.executor);
   }
 
   private FirebaseOptions buildOptions(ThreadManager threadManager) {
@@ -225,21 +242,24 @@ public class ThreadManagerTest {
         .build();
   }
 
-  public static class MockThreadManager extends ThreadManager {
+  private static class MockThreadManager extends ThreadManager {
 
     private final List<Event> events = new ArrayList<>();
+    private final ExecutorService executor;
+
+    MockThreadManager(ExecutorService executor) {
+      this.executor = executor;
+    }
 
     @Override
-    protected ScheduledExecutorService getExecutor(@NonNull FirebaseApp app) {
+    protected ExecutorService getExecutor(@NonNull FirebaseApp app) {
       events.add(new Event(Event.TYPE_GET_EXECUTOR, app, null));
-      return Executors.newSingleThreadScheduledExecutor();
+      return executor;
     }
 
     @Override
-    protected void releaseExecutor(@NonNull FirebaseApp app,
-        @NonNull ScheduledExecutorService executor) {
+    protected void releaseExecutor(@NonNull FirebaseApp app, @NonNull ExecutorService executor) {
       events.add(new Event(Event.TYPE_RELEASE_EXECUTOR, app, executor));
-      executor.shutdownNow();
     }
 
     @Override
@@ -256,9 +276,9 @@ public class ThreadManagerTest {
 
     private final int type;
     private final FirebaseApp app;
-    private ScheduledExecutorService executor;
+    private ExecutorService executor;
 
-    public Event(int type, @Nullable FirebaseApp app, @Nullable ScheduledExecutorService executor) {
+    public Event(int type, @Nullable FirebaseApp app, @Nullable ExecutorService executor) {
       this.type = type;
       this.app = app;
       this.executor = executor;
diff --git a/src/test/java/com/google/firebase/database/core/JvmAuthTokenProviderTest.java b/src/test/java/com/google/firebase/database/core/JvmAuthTokenProviderTest.java
new file mode 100644
index 0000000..a174b3e
--- /dev/null
+++ b/src/test/java/com/google/firebase/database/core/JvmAuthTokenProviderTest.java
@@ -0,0 +1,277 @@
+/*
+ * Copyright 2017 Google Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.firebase.database.core;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.cedarsoftware.util.DeepEquals;
+import com.google.auth.oauth2.AccessToken;
+import com.google.auth.oauth2.OAuth2Credentials;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.firebase.FirebaseApp;
+import com.google.firebase.FirebaseOptions;
+import com.google.firebase.TestOnlyImplFirebaseTrampolines;
+import com.google.firebase.auth.MockGoogleCredentials;
+import com.google.firebase.testing.TestUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.json.JSONObject;
+import org.junit.After;
+import org.junit.Test;
+
+public class JvmAuthTokenProviderTest {
+
+  private static final Executor DIRECT_EXECUTOR = MoreExecutors.directExecutor();
+
+  @After
+  public void tearDown() {
+    TestOnlyImplFirebaseTrampolines.clearInstancesForTest();
+  }
+
+  @Test
+  public void testGetToken() throws IOException, InterruptedException {
+    MockGoogleCredentials credentials = new MockGoogleCredentials("mock-token");
+    TokenRefreshDetector refreshDetector = new TokenRefreshDetector();
+    credentials.addChangeListener(refreshDetector);
+    credentials.refresh();
+    assertEquals(1, refreshDetector.count);
+
+    FirebaseOptions options = new FirebaseOptions.Builder()
+        .setCredentials(credentials)
+        .build();
+    FirebaseApp app = FirebaseApp.initializeApp(options);
+
+    JvmAuthTokenProvider provider = new JvmAuthTokenProvider(app, DIRECT_EXECUTOR);
+    TestGetTokenListener listener = new TestGetTokenListener();
+    provider.getToken(true, listener);
+    assertToken(listener.get(), "mock-token", ImmutableMap.<String, Object>of());
+    assertEquals(2, refreshDetector.count);
+  }
+
+  @Test
+  public void testGetTokenNoRefresh() throws IOException, InterruptedException {
+    MockGoogleCredentials credentials = new MockGoogleCredentials("mock-token");
+    TokenRefreshDetector refreshDetector = new TokenRefreshDetector();
+    credentials.addChangeListener(refreshDetector);
+    credentials.refresh();
+    assertEquals(1, refreshDetector.count);
+
+    FirebaseOptions options = new FirebaseOptions.Builder()
+        .setCredentials(credentials)
+        .build();
+    FirebaseApp app = FirebaseApp.initializeApp(options);
+
+    JvmAuthTokenProvider provider = new JvmAuthTokenProvider(app, DIRECT_EXECUTOR);
+    TestGetTokenListener listener = new TestGetTokenListener();
+    provider.getToken(false, listener);
+    assertToken(listener.get(), "mock-token", ImmutableMap.<String, Object>of());
+    assertEquals(1, refreshDetector.count);
+  }
+
+  @Test
+  public void testGetTokenWithAuthOverrides() throws InterruptedException {
+    MockGoogleCredentials credentials = new MockGoogleCredentials("mock-token");
+    Map<String, Object> auth = ImmutableMap.<String, Object>of("uid", "test");
+    FirebaseOptions options = new FirebaseOptions.Builder()
+        .setCredentials(credentials)
+        .setDatabaseAuthVariableOverride(auth)
+        .build();
+    FirebaseApp app = FirebaseApp.initializeApp(options);
+
+    JvmAuthTokenProvider provider = new JvmAuthTokenProvider(app, DIRECT_EXECUTOR);
+    TestGetTokenListener listener = new TestGetTokenListener();
+    provider.getToken(true, listener);
+    assertToken(listener.get(), "mock-token", auth);
+  }
+
+  @Test
+  public void testGetTokenError() throws InterruptedException {
+    MockGoogleCredentials credentials = new MockGoogleCredentials("mock-token") {
+      @Override
+      public AccessToken refreshAccessToken() throws IOException {
+        throw new RuntimeException("Test error");
+      }
+    };
+    FirebaseOptions options = new FirebaseOptions.Builder()
+        .setCredentials(credentials)
+        .build();
+    FirebaseApp app = FirebaseApp.initializeApp(options);
+
+    JvmAuthTokenProvider provider = new JvmAuthTokenProvider(app, DIRECT_EXECUTOR);
+    TestGetTokenListener listener = new TestGetTokenListener();
+    provider.getToken(true, listener);
+    assertEquals("java.lang.RuntimeException: Test error", listener.get());
+  }
+
+  @Test
+  public void testAddTokenChangeListener() throws IOException {
+    final AtomicInteger counter = new AtomicInteger(0);
+    MockGoogleCredentials credentials = new MockGoogleCredentials() {
+      @Override
+      public AccessToken refreshAccessToken() throws IOException {
+        Date expiry = new Date(System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1));
+        return new AccessToken("token-" + counter.getAndIncrement(), expiry);
+      }
+    };
+
+    FirebaseOptions options = new FirebaseOptions.Builder()
+        .setCredentials(credentials)
+        .build();
+    FirebaseApp app = FirebaseApp.initializeApp(options);
+
+    // Disable proactive token refresh, so only explicit refresh events are in play.
+    JvmAuthTokenProvider provider = new JvmAuthTokenProvider(app, DIRECT_EXECUTOR, false);
+    final List<String> tokens = new ArrayList<>();
+    provider.addTokenChangeListener(new AuthTokenProvider.TokenChangeListener() {
+      @Override
+      public void onTokenChange(String token) {
+        tokens.add(token);
+      }
+    });
+
+    for (int i = 0; i < 10; i++) {
+      // Each refresh event should notify the TokenChangeListener. And since we are using a
+      // direct executor, the notification fires on the same thread in a blocking manner.
+      credentials.refresh();
+      assertEquals(i + 1, tokens.size());
+      assertToken(tokens.get(i), "token-" + i, ImmutableMap.<String, Object>of());
+    }
+  }
+
+  @Test
+  public void testTokenChangeListenerThread() throws InterruptedException, IOException {
+    MockGoogleCredentials credentials = new MockGoogleCredentials();
+    FirebaseOptions options = new FirebaseOptions.Builder()
+        .setCredentials(credentials)
+        .build();
+    FirebaseApp app = FirebaseApp.initializeApp(options);
+
+    // Disable proactive token refresh, so only explicit refresh events are in play.
+    ThreadFactory threadFactory = new ThreadFactoryBuilder()
+        .setNameFormat("auth-token-provider-thread")
+        .setDaemon(true)
+        .build();
+    ExecutorService executor = Executors.newSingleThreadExecutor(threadFactory);
+    try {
+      JvmAuthTokenProvider provider = new JvmAuthTokenProvider(app, executor, false);
+
+      final AtomicReference<String> result = new AtomicReference<>();
+      final Semaphore semaphore = new Semaphore(0);
+      provider.addTokenChangeListener(new AuthTokenProvider.TokenChangeListener() {
+        @Override
+        public void onTokenChange(String token) {
+          result.set(Thread.currentThread().getName());
+          semaphore.release();
+        }
+      });
+
+      credentials.refresh();
+      assertTrue(semaphore.tryAcquire(TestUtils.TEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
+      assertEquals("auth-token-provider-thread", result.get());
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  public void testTokenAutoRefresh() throws InterruptedException {
+    MockGoogleCredentials credentials = new MockGoogleCredentials();
+    final Semaphore semaphore = new Semaphore(0);
+    credentials.addChangeListener(new OAuth2Credentials.CredentialsChangedListener() {
+      @Override
+      public void onChanged(OAuth2Credentials credentials) throws IOException {
+        semaphore.release();
+      }
+    });
+
+    FirebaseOptions options = new FirebaseOptions.Builder()
+        .setCredentials(credentials)
+        .build();
+    FirebaseApp app = FirebaseApp.initializeApp(options);
+
+    // Creating JvmAuthTokenProvider should start the proactive token refresher, which should
+    // immediately refresh the credential once.
+    new JvmAuthTokenProvider(app, DIRECT_EXECUTOR);
+    assertTrue(semaphore.tryAcquire(TestUtils.TEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
+  }
+
+  private void assertToken(String token, String expectedToken, Map<String, Object> expectedAuth) {
+    assertTrue(token.startsWith("gauth|"));
+    String jsonString = token.substring(6);
+    JSONObject json = new JSONObject(jsonString);
+    assertEquals(expectedToken, json.getString("token"));
+
+    Map<String, Object> auth = json.getJSONObject("auth").toMap();
+    DeepEquals.deepEquals(expectedAuth, auth);
+  }
+
+  private static class TestGetTokenListener
+      implements AuthTokenProvider.GetTokenCompletionListener {
+
+    private final Semaphore semaphore = new Semaphore(0);
+    private final AtomicReference<String> result = new AtomicReference<>(null);
+
+    @Override
+    public void onSuccess(String token) {
+      result.set(token);
+      semaphore.release();
+    }
+
+    @Override
+    public void onError(String error) {
+      result.set(error);
+      semaphore.release();
+    }
+
+    String get() throws InterruptedException {
+      if (semaphore.tryAcquire(TestUtils.TEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
+        return result.get();
+      }
+      fail("Timed out while waiting for GetTokenCompletionListener");
+      return null;
+    }
+  }
+
+  private static class TokenRefreshDetector
+      implements OAuth2Credentials.CredentialsChangedListener {
+
+    private int count = 0;
+
+    @Override
+    public void onChanged(OAuth2Credentials credentials) throws IOException {
+      count++;
+    }
+  }
+
+}
diff --git a/src/test/java/com/google/firebase/database/core/JvmPlatformTest.java b/src/test/java/com/google/firebase/database/core/JvmPlatformTest.java
index 37ad6de..c691291 100644
--- a/src/test/java/com/google/firebase/database/core/JvmPlatformTest.java
+++ b/src/test/java/com/google/firebase/database/core/JvmPlatformTest.java
@@ -26,8 +26,9 @@ import com.google.firebase.database.FirebaseDatabase;
 import com.google.firebase.internal.NonNull;
 import com.google.firebase.testing.ServiceAccount;
 import com.google.firebase.testing.TestUtils;
+
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Assert;
@@ -40,13 +41,13 @@ public class JvmPlatformTest {
     final AtomicInteger count = new AtomicInteger(0);
     ThreadManager threadManager = new ThreadManager() {
       @Override
-      protected ScheduledExecutorService getExecutor(@NonNull FirebaseApp app) {
-        return Executors.newSingleThreadScheduledExecutor();
+      protected ExecutorService getExecutor(@NonNull FirebaseApp app) {
+        return Executors.newSingleThreadExecutor();
       }
 
       @Override
       protected void releaseExecutor(@NonNull FirebaseApp app,
-          @NonNull ScheduledExecutorService executor) {
+          @NonNull ExecutorService executor) {
       }
 
       @Override
@@ -66,12 +67,12 @@ public class JvmPlatformTest {
       Context cfg = new DatabaseConfig();
       cfg.firebaseApp = app;
       cfg.freeze();
-      // EventTarget and RunLoop
-      assertEquals(2, count.get());
+      // EventTarget, RunLoop and AuthTokenProvider (token refresher)
+      assertEquals(3, count.get());
 
       cfg.getConnectionContext();
       // ConnectionContext which gets passed to all low-level socket management code.
-      assertEquals(3, count.get());
+      assertEquals(4, count.get());
     } finally {
       TestOnlyImplFirebaseTrampolines.clearInstancesForTest();
     }
diff --git a/src/test/java/com/google/firebase/internal/FirebaseExecutorsTest.java b/src/test/java/com/google/firebase/internal/FirebaseThreadManagersTest.java
similarity index 87%
rename from src/test/java/com/google/firebase/internal/FirebaseExecutorsTest.java
rename to src/test/java/com/google/firebase/internal/FirebaseThreadManagersTest.java
index d8e7792..7a0d2d1 100644
--- a/src/test/java/com/google/firebase/internal/FirebaseExecutorsTest.java
+++ b/src/test/java/com/google/firebase/internal/FirebaseThreadManagersTest.java
@@ -28,19 +28,20 @@ import com.google.firebase.FirebaseOptions;
 import com.google.firebase.ImplFirebaseTrampolines;
 import com.google.firebase.TestOnlyImplFirebaseTrampolines;
 import com.google.firebase.auth.MockGoogleCredentials;
-import com.google.firebase.internal.FirebaseExecutors.GlobalThreadManager;
+import com.google.firebase.internal.FirebaseThreadManagers.GlobalThreadManager;
 import com.google.firebase.tasks.Tasks;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import org.junit.After;
+import org.junit.Assume;
 import org.junit.Test;
 
-public class FirebaseExecutorsTest {
+public class FirebaseThreadManagersTest {
 
   @After
   public void tearDown() {
@@ -57,11 +58,11 @@ public class FirebaseExecutorsTest {
     FirebaseApp defaultApp = FirebaseApp.initializeApp(options);
     assertEquals(1, threadManager.initCount);
 
-    ScheduledExecutorService exec1 = threadManager.getExecutor(defaultApp);
+    ExecutorService exec1 = threadManager.getExecutor(defaultApp);
     assertEquals(1, threadManager.initCount);
     assertFalse(exec1.isShutdown());
 
-    ScheduledExecutorService exec2 = threadManager.getExecutor(defaultApp);
+    ExecutorService exec2 = threadManager.getExecutor(defaultApp);
     assertEquals(1, threadManager.initCount);
     assertFalse(exec2.isShutdown());
 
@@ -82,8 +83,8 @@ public class FirebaseExecutorsTest {
     FirebaseApp customApp = FirebaseApp.initializeApp(options, "customApp");
     assertEquals(0, threadManager.initCount);
 
-    ScheduledExecutorService exec1 = threadManager.getExecutor(defaultApp);
-    ScheduledExecutorService exec2 = threadManager.getExecutor(customApp);
+    ExecutorService exec1 = threadManager.getExecutor(defaultApp);
+    ExecutorService exec2 = threadManager.getExecutor(customApp);
     assertEquals(1, threadManager.initCount);
     assertFalse(exec1.isShutdown());
 
@@ -107,7 +108,7 @@ public class FirebaseExecutorsTest {
     FirebaseApp defaultApp = FirebaseApp.initializeApp(options);
     assertEquals(1, threadManager.initCount);
 
-    ScheduledExecutorService exec1 = threadManager.getExecutor(defaultApp);
+    ExecutorService exec1 = threadManager.getExecutor(defaultApp);
     assertEquals(1, threadManager.initCount);
     assertFalse(exec1.isShutdown());
 
@@ -116,7 +117,7 @@ public class FirebaseExecutorsTest {
     assertTrue(exec1.isShutdown());
 
     // Simulate app re-init
-    ScheduledExecutorService exec2 = threadManager.getExecutor(defaultApp);
+    ExecutorService exec2 = threadManager.getExecutor(defaultApp);
     assertEquals(2, threadManager.initCount);
     assertNotSame(exec1, exec2);
 
@@ -126,6 +127,8 @@ public class FirebaseExecutorsTest {
 
   @Test
   public void testDefaultThreadManager() throws Exception {
+    Assume.assumeFalse(GaeThreadFactory.isAvailable());
+
     FirebaseOptions options = new FirebaseOptions.Builder()
         .setCredentials(new MockGoogleCredentials())
         .build();
@@ -160,13 +163,13 @@ public class FirebaseExecutorsTest {
     private int initCount = 0;
 
     @Override
-    protected ScheduledExecutorService doInit() {
+    protected ExecutorService doInit() {
       initCount++;
-      return Executors.newSingleThreadScheduledExecutor();
+      return Executors.newSingleThreadExecutor();
     }
 
     @Override
-    protected void doCleanup(ScheduledExecutorService executorService) {
+    protected void doCleanup(ExecutorService executorService) {
       executorService.shutdownNow();
     }