address comments

Author: mutianf

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableClientContext.java

@@ -19,9 +19,8 @@
 import com.google.api.core.InternalApi;
 import com.google.api.gax.core.BackgroundResource;
 import com.google.api.gax.core.CredentialsProvider;
-import com.google.api.gax.core.ExecutorAsBackgroundResource;
+import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
-import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.rpc.ClientContext;
 import com.google.auth.Credentials;
@@ -36,7 +35,6 @@
 import com.google.cloud.bigtable.data.v2.stub.metrics.NoopMetricsProvider;
 import com.google.cloud.bigtable.gaxx.grpc.BigtableTransportChannelProvider;
 import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer;
-import com.google.common.collect.ImmutableList;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.opentelemetry.GrpcOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
@@ -62,6 +60,9 @@ public class BigtableClientContext {
   @Nullable private final OpenTelemetrySdk internalOpenTelemetry;
   private final MetricsProvider metricsProvider;
   private final ClientContext clientContext;
+  // the background executor shared for OTEL instances and monitoring client and all other
+  // background tasks
+  private final ExecutorProvider backgroundExecutorProvider;
 
   public static BigtableClientContext create(EnhancedBigtableStubSettings settings)
       throws IOException {
@@ -79,16 +80,12 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
 
     String universeDomain = settings.getUniverseDomain();
 
-    boolean canAutoCloseExecutor = true;
+    boolean shouldAutoClose = settings.getBackgroundExecutorProvider().shouldAutoClose();
     ScheduledExecutorService backgroundExecutor =
         settings.getBackgroundExecutorProvider().getExecutor();
-    if (settings.getBackgroundExecutorProvider() instanceof FixedExecutorProvider) {
-      // if the background executor in the settings is already a FixedExecutorProvider,
-      // we can't assume that we can autoclose it and the life cycle should be managed
-      // by the application
-      canAutoCloseExecutor = false;
-    }
-    FixedExecutorProvider executorProvider = FixedExecutorProvider.create(backgroundExecutor);
+    // TODO: after gax change is merged, migrate to use gax's FixedExecutorProvider
+    BigtableExecutorProvider executorProvider =
+        BigtableExecutorProvider.create(backgroundExecutor, shouldAutoClose);
     builder.setBackgroundExecutorProvider(executorProvider);
 
     // Set up OpenTelemetry
@@ -162,23 +159,16 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
     }
 
     ClientContext clientContext = ClientContext.create(builder.build());
-    if (canAutoCloseExecutor) {
-      // Since we converted background executor to a FixedExecutorProvider, we need
-      // to add it back to the background resources, so it will be closed when we close the
-      // client context.
-      ImmutableList<BackgroundResource> backgroundResources =
-          ImmutableList.<BackgroundResource>builder()
-              .addAll(clientContext.getBackgroundResources())
-              .add(new ExecutorAsBackgroundResource(backgroundExecutor))
-              .build();
-      clientContext = clientContext.toBuilder().setBackgroundResources(backgroundResources).build();
-    }
     if (channelPoolMetricsTracer != null) {
       channelPoolMetricsTracer.start(clientContext.getExecutor());
     }
 
     return new BigtableClientContext(
-        clientContext, openTelemetry, internalOtel, settings.getMetricsProvider());
+        clientContext,
+        openTelemetry,
+        internalOtel,
+        settings.getMetricsProvider(),
+        executorProvider);
   }
 
   private static void configureGrpcOtel(
@@ -212,11 +202,13 @@ private BigtableClientContext(
       ClientContext clientContext,
       @Nullable OpenTelemetry openTelemetry,
       @Nullable OpenTelemetrySdk internalOtel,
-      MetricsProvider metricsProvider) {
+      MetricsProvider metricsProvider,
+      ExecutorProvider backgroundExecutorProvider) {
     this.clientContext = clientContext;
     this.openTelemetry = openTelemetry;
     this.internalOpenTelemetry = internalOtel;
     this.metricsProvider = metricsProvider;
+    this.backgroundExecutorProvider = backgroundExecutorProvider;
   }
 
   public OpenTelemetry getOpenTelemetry() {
@@ -229,7 +221,11 @@ public ClientContext getClientContext() {
 
   public BigtableClientContext withClientContext(ClientContext clientContext) {
     return new BigtableClientContext(
-        clientContext, openTelemetry, internalOpenTelemetry, metricsProvider);
+        clientContext,
+        openTelemetry,
+        internalOpenTelemetry,
+        metricsProvider,
+        backgroundExecutorProvider);
   }
 
   public void close() throws Exception {
@@ -242,6 +238,9 @@ public void close() throws Exception {
     if (metricsProvider instanceof DefaultMetricsProvider && openTelemetry != null) {
       ((OpenTelemetrySdk) openTelemetry).close();
     }
+    if (backgroundExecutorProvider.shouldAutoClose()) {
+      backgroundExecutorProvider.getExecutor().shutdownNow();
+    }
   }
 
   private static OpenTelemetry getOpenTelemetryFromMetricsProvider(

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableExecutorProvider.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * 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
+ *
+ *     https://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.cloud.bigtable.data.v2.stub;
+
+import com.google.api.gax.core.ExecutorProvider;
+import java.util.concurrent.ScheduledExecutorService;
+
+// TODO: migrate to gax's FixedExecutorProvider once the change is merged
+class BigtableExecutorProvider implements ExecutorProvider {
+
+  private final ScheduledExecutorService executorService;
+  private final boolean shouldAutoClose;
+
+  @Override
+  public boolean shouldAutoClose() {
+    return shouldAutoClose;
+  }
+
+  @Override
+  public ScheduledExecutorService getExecutor() {
+    return executorService;
+  }
+
+  static BigtableExecutorProvider create(
+      ScheduledExecutorService executor, boolean shouldAutoClose) {
+    return new BigtableExecutorProvider(executor, shouldAutoClose);
+  }
+
+  private BigtableExecutorProvider(
+      ScheduledExecutorService executorService, boolean shouldAutoClose) {
+    this.shouldAutoClose = shouldAutoClose;
+    this.executorService = executorService;
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableCloudMonitoringExporter.java

@@ -132,7 +132,9 @@ static BigtableCloudMonitoringExporter create(
 
     // If background executor is not null, use it for the monitoring client. This allows us to
     // share the same background executor with the data client. When it's null, the monitoring
-    // client will create a new executor service from InstantiatingExecutorProvider.
+    // client will create a new executor service from InstantiatingExecutorProvider. It could be
+    // null if someone uses a CustomOpenTelemetryMetricsProvider#setupSdkMeterProvider without
+    // the executor.
     if (executorService != null) {
       settingsBuilder.setBackgroundExecutorProvider(FixedExecutorProvider.create(executorService));
     }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/CustomOpenTelemetryMetricsProvider.java

@@ -71,31 +71,31 @@ public OpenTelemetry getOpenTelemetry() {
    * Convenient method to set up SdkMeterProviderBuilder with the default credential and endpoint.
    */
   public static void setupSdkMeterProvider(SdkMeterProviderBuilder builder) throws IOException {
-    setupSdkMeterProvider(builder, null, null);
+    setupSdkMeterProvider(builder, null, null, null);
   }
 
   /** Convenient method to set up SdkMeterProviderBuilder with a custom credential. */
   public static void setupSdkMeterProvider(SdkMeterProviderBuilder builder, Credentials credentials)
       throws IOException {
-    setupSdkMeterProvider(builder, credentials, null);
+    setupSdkMeterProvider(builder, credentials, null, null);
   }
 
   /** Convenient method to set up SdkMeterProviderBuilder with a custom endpoint. */
   public static void setupSdkMeterProvider(SdkMeterProviderBuilder builder, String endpoint)
       throws IOException {
-    setupSdkMeterProvider(builder, null, endpoint);
+    setupSdkMeterProvider(builder, null, endpoint, null);
   }
 
   /** Convenient method to set up SdkMeterProviderBuilder with custom credentials and endpoint. */
   public static void setupSdkMeterProvider(
       SdkMeterProviderBuilder builder, Credentials credentials, String endpoint)
       throws IOException {
-    BuiltinMetricsView.registerBuiltinMetrics(credentials, builder, endpoint);
+    setupSdkMeterProvider(builder, credentials, endpoint, null);
   }
 
   /**
-   * Convenient method to set up SdkMeterProviderBuilder with custom credentials, endpoint and
-   * executor service.
+   * Convenient method to set up SdkMeterProviderBuilder with custom credentials, endpoint and a
+   * shared executor service.
    */
   public static void setupSdkMeterProvider(
       SdkMeterProviderBuilder builder,