fix: Watchdog controls lifecycle of the future, not executor (#1890)

Author: meltsufin

gax/src/main/java/com/google/api/gax/rpc/Watchdog.java

@@ -36,9 +36,11 @@
 import java.util.Map.Entry;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nonnull;
@@ -61,6 +63,7 @@
  * </ul>
  */
 public final class Watchdog implements Runnable, BackgroundResource {
+
  private static final Logger LOG = Logger.getLogger(Watchdog.class.getName());
 
  // Dummy value to convert the ConcurrentHashMap into a Set
@@ -138,12 +141,12 @@ public void shutdown() {
 
  @Override
  public boolean isShutdown() {
- return executor.isShutdown();
+ return future.isCancelled();
  }
 
  @Override
  public boolean isTerminated() {
- return executor.isTerminated();
+ return future.isDone();
  }
 
  @Override
@@ -153,7 +156,14 @@ public void shutdownNow() {
 
  @Override
  public boolean awaitTermination(long duration, TimeUnit unit) throws InterruptedException {
- return executor.awaitTermination(duration, unit);
+ try {
+ future.get(duration, unit);
+ return true;
+ } catch (ExecutionException | CancellationException e) {
+ return true;
+ } catch (TimeoutException e) {
+ return false;
+ }
  }
 
  @Override

gax/src/main/java/com/google/api/gax/rpc/WatchdogProvider.java

@@ -49,5 +49,6 @@ public interface WatchdogProvider {
 
  Watchdog getWatchdog();
 
+ /** Return true if the watchdog should be automatically unscheduled. */
  boolean shouldAutoClose();
 }