fix: [Approach 1]Watchdog does not shut down executor on client closing if the executor is provided by ExecutorProvider.

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

@@ -218,7 +218,7 @@ public static ClientContext create(StubSettings settings) throws IOException {
         watchdogProvider = watchdogProvider.withClock(clock);
       }
       if (watchdogProvider.needsExecutor()) {
-        watchdogProvider = watchdogProvider.withExecutor(backgroundExecutor);
+        watchdogProvider = watchdogProvider.withExecutor(backgroundExecutor).withAutoClose(false);
       }
       watchdog = watchdogProvider.getWatchdog();
     }

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

@@ -48,18 +48,21 @@ public final class InstantiatingWatchdogProvider implements WatchdogProvider {
   @Nullable private final ApiClock clock;
   @Nullable private final ScheduledExecutorService executor;
   @Nullable private final Duration checkInterval;
+  private final boolean autoClose;
 
   public static WatchdogProvider create() {
-    return new InstantiatingWatchdogProvider(null, null, null);
+    return new InstantiatingWatchdogProvider(null, null, null, true);
   }
 
   private InstantiatingWatchdogProvider(
       @Nullable ApiClock clock,
       @Nullable ScheduledExecutorService executor,
-      @Nullable Duration checkInterval) {
+      @Nullable Duration checkInterval,
+      boolean autoClose) {
     this.clock = clock;
     this.executor = executor;
     this.checkInterval = checkInterval;
+    this.autoClose = autoClose;
   }
 
   @Override
@@ -70,7 +73,7 @@ public boolean needsClock() {
   @Override
   public WatchdogProvider withClock(@Nonnull ApiClock clock) {
     return new InstantiatingWatchdogProvider(
-        Preconditions.checkNotNull(clock), executor, checkInterval);
+        Preconditions.checkNotNull(clock), executor, checkInterval, autoClose);
   }
 
   @Override
@@ -81,7 +84,7 @@ public boolean needsCheckInterval() {
   @Override
   public WatchdogProvider withCheckInterval(@Nonnull Duration checkInterval) {
     return new InstantiatingWatchdogProvider(
-        clock, executor, Preconditions.checkNotNull(checkInterval));
+        clock, executor, Preconditions.checkNotNull(checkInterval), autoClose);
   }
 
   @Override
@@ -92,7 +95,7 @@ public boolean needsExecutor() {
   @Override
   public WatchdogProvider withExecutor(ScheduledExecutorService executor) {
     return new InstantiatingWatchdogProvider(
-        clock, Preconditions.checkNotNull(executor), checkInterval);
+        clock, Preconditions.checkNotNull(executor), checkInterval, autoClose);
   }
 
   @SuppressWarnings("ConstantConditions")
@@ -111,8 +114,13 @@ public Watchdog getWatchdog() {
     return Watchdog.create(clock, checkInterval, executor);
   }
 
+  @Override
+  public WatchdogProvider withAutoClose(boolean autoClose) {
+    return new InstantiatingWatchdogProvider(clock, executor, checkInterval, autoClose);
+  }
+
   @Override
   public boolean shouldAutoClose() {
-    return true;
+    return autoClose;
   }
 }

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

@@ -134,6 +134,7 @@ private void runUnsafe() {
   @Override
   public void shutdown() {
     future.cancel(false);
+    executor.shutdown();
   }
 
   @Override
@@ -149,6 +150,7 @@ public boolean isTerminated() {
   @Override
   public void shutdownNow() {
     future.cancel(true);
+    executor.shutdownNow();
   }
 
   @Override

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

@@ -30,6 +30,7 @@
 package com.google.api.gax.rpc;
 
 import com.google.api.core.ApiClock;
+import com.google.api.core.BetaApi;
 import java.util.concurrent.ScheduledExecutorService;
 import javax.annotation.Nonnull;
 import org.threeten.bp.Duration;
@@ -49,5 +50,13 @@ public interface WatchdogProvider {
 
   Watchdog getWatchdog();
 
+  @BetaApi
+  default WatchdogProvider withAutoClose(boolean autoClose) {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
+  /**
+   * If this is true, closing the client will automatically shut down the executor used by Watchdog
+   */
   boolean shouldAutoClose();
 }

gax/src/test/java/com/google/api/gax/rpc/WatchdogTest.java

@@ -213,10 +213,11 @@ public void testWatchdogBeingClosed() {
         .scheduleAtFixedRate(
             underTest, checkInterval.toMillis(), checkInterval.toMillis(), TimeUnit.MILLISECONDS);
     Mockito.verify(future, Mockito.times(2)).cancel(false);
+    Mockito.verify(mockExecutor, Mockito.times(2)).shutdown();
 
     underTest.shutdownNow();
     Mockito.verify(future).cancel(true);
-    Mockito.verifyNoMoreInteractions(mockExecutor);
+    Mockito.verify(mockExecutor).shutdownNow();
   }
 
   static class AccumulatingObserver<T> implements ResponseObserver<T> {