fix: use the same background executor in otel reader and monitoring c…

google-cloud-bigtable/clirr-ignored-differences.xml

@@ -499,4 +499,16 @@
         <className>com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPoolSettings$Builder</className>
         <method>com.google.cloud.bigtable.gaxx.grpc.BigtableChannelPoolSettings$Builder setLoadBalancingStrategy(com.google.cloud.bigtable.gaxx.grpc.BigtableChannelPoolSettings$LoadBalancingStrategy)</method>
     </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7004</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings$InternalMetricsProvider</className>
+        <method>*</method>
+    </difference>
+    <difference>
+        <!-- InternalApi was updated -->
+        <differenceType>7004</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/metrics/Util</className>
+        <method>*</method>
+    </difference>
 </differences>

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

@@ -19,6 +19,7 @@
 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.ExecutorProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.rpc.ClientContext;
@@ -41,6 +42,7 @@
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -58,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 {
@@ -75,6 +80,14 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
 
     String universeDomain = settings.getUniverseDomain();
 
+    boolean shouldAutoClose = settings.getBackgroundExecutorProvider().shouldAutoClose();
+    ScheduledExecutorService backgroundExecutor =
+        settings.getBackgroundExecutorProvider().getExecutor();
+    // TODO: after gax change is merged, migrate to use gax's FixedExecutorProvider
+    BigtableExecutorProvider executorProvider =
+        BigtableExecutorProvider.create(backgroundExecutor, shouldAutoClose);
+    builder.setBackgroundExecutorProvider(executorProvider);
+
     // Set up OpenTelemetry
     OpenTelemetry openTelemetry = null;
     try {
@@ -85,7 +98,8 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
               settings.getMetricsProvider(),
               credentials,
               settings.getMetricsEndpoint(),
-              universeDomain);
+              universeDomain,
+              backgroundExecutor);
     } catch (Throwable t) {
       logger.log(Level.WARNING, "Failed to get OTEL, will skip exporting client side metrics", t);
     }
@@ -103,7 +117,9 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
     // no reason to build the internal OtelProvider
     if (transportProvider != null) {
       internalOtel =
-          settings.getInternalMetricsProvider().createOtelProvider(settings, credentials);
+          settings
+              .getInternalMetricsProvider()
+              .createOtelProvider(settings, credentials, backgroundExecutor);
       if (internalOtel != null) {
         channelPoolMetricsTracer =
             new ChannelPoolMetricsTracer(
@@ -148,7 +164,11 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
     }
 
     return new BigtableClientContext(
-        clientContext, openTelemetry, internalOtel, settings.getMetricsProvider());
+        clientContext,
+        openTelemetry,
+        internalOtel,
+        settings.getMetricsProvider(),
+        executorProvider);
   }
 
   private static void configureGrpcOtel(
@@ -182,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() {
@@ -199,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 {
@@ -212,13 +238,17 @@ public void close() throws Exception {
     if (metricsProvider instanceof DefaultMetricsProvider && openTelemetry != null) {
       ((OpenTelemetrySdk) openTelemetry).close();
     }
+    if (backgroundExecutorProvider.shouldAutoClose()) {
+      backgroundExecutorProvider.getExecutor().shutdown();
+    }
   }
 
   private static OpenTelemetry getOpenTelemetryFromMetricsProvider(
       MetricsProvider metricsProvider,
       @Nullable Credentials defaultCredentials,
       @Nullable String metricsEndpoint,
-      String universeDomain)
+      String universeDomain,
+      ScheduledExecutorService executor)
       throws IOException {
     if (metricsProvider instanceof CustomOpenTelemetryMetricsProvider) {
       CustomOpenTelemetryMetricsProvider customMetricsProvider =
@@ -230,7 +260,8 @@ private static OpenTelemetry getOpenTelemetryFromMetricsProvider(
               ? BigtableDataSettings.getMetricsCredentials()
               : defaultCredentials;
       DefaultMetricsProvider defaultMetricsProvider = (DefaultMetricsProvider) metricsProvider;
-      return defaultMetricsProvider.getOpenTelemetry(metricsEndpoint, universeDomain, credentials);
+      return defaultMetricsProvider.getOpenTelemetry(
+          metricsEndpoint, universeDomain, credentials, executor);
     } else if (metricsProvider instanceof NoopMetricsProvider) {
       return null;
     }

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/EnhancedBigtableStubSettings.java

@@ -69,6 +69,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Logger;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -1388,11 +1389,14 @@ public String toString() {
   public interface InternalMetricsProvider {
     @Nullable
     OpenTelemetrySdk createOtelProvider(
-        EnhancedBigtableStubSettings userSettings, Credentials creds) throws IOException;
+        EnhancedBigtableStubSettings userSettings,
+        Credentials creds,
+        ScheduledExecutorService executor)
+        throws IOException;
   }
 
   private static final InternalMetricsProvider DEFAULT_INTERNAL_OTEL_PROVIDER =
       Util::newInternalOpentelemetry;
   private static final InternalMetricsProvider DISABLED_INTERNAL_OTEL_PROVIDER =
-      (ignored1, ignored2) -> null;
+      (ignored1, ignored2, ignored3) -> null;
 }

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

@@ -35,6 +35,7 @@
 import com.google.api.core.InternalApi;
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
+import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.core.NoCredentialsProvider;
 import com.google.api.gax.rpc.PermissionDeniedException;
 import com.google.auth.Credentials;
@@ -65,6 +66,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
@@ -115,7 +117,8 @@ static BigtableCloudMonitoringExporter create(
       @Nullable Credentials credentials,
       @Nullable String endpoint,
       String universeDomain,
-      TimeSeriesConverter converter)
+      TimeSeriesConverter converter,
+      @Nullable ScheduledExecutorService executorService)
       throws IOException {
     Preconditions.checkNotNull(universeDomain);
     MetricServiceSettings.Builder settingsBuilder = MetricServiceSettings.newBuilder();
@@ -127,6 +130,15 @@ static BigtableCloudMonitoringExporter create(
 
     settingsBuilder.setUniverseDomain(universeDomain);
 
+    // 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. It could be
+    // null if someone uses a CustomOpenTelemetryMetricsProvider#setupSdkMeterProvider without
+    // the executor.
+    if (executorService != null) {
+      settingsBuilder.setBackgroundExecutorProvider(FixedExecutorProvider.create(executorService));
+    }
+
     if (MONITORING_ENDPOINT_OVERRIDE_SYS_PROP != null) {
       logger.warning(
           "Setting the monitoring endpoint through system variable will be removed in future"

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

@@ -22,8 +22,10 @@
 import io.opentelemetry.sdk.metrics.View;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReaderBuilder;
 import java.io.IOException;
 import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
 import javax.annotation.Nullable;
 
 /**
@@ -100,27 +102,47 @@ public static void registerBuiltinMetrics(
       @Nullable Credentials credentials, SdkMeterProviderBuilder builder, @Nullable String endpoint)
       throws IOException {
     registerBuiltinMetricsWithUniverseDomain(
-        credentials, builder, endpoint, Credentials.GOOGLE_DEFAULT_UNIVERSE);
+        credentials, builder, endpoint, Credentials.GOOGLE_DEFAULT_UNIVERSE, null);
+  }
+
+  /**
+   * Register built-in metrics on the {@link SdkMeterProviderBuilder} with custom credentials,
+   * endpoint and executor service.
+   */
+  public static void registerBuiltinMetrics(
+      @Nullable Credentials credentials,
+      SdkMeterProviderBuilder builder,
+      @Nullable String endpoint,
+      @Nullable ScheduledExecutorService executorService)
+      throws IOException {
+    registerBuiltinMetricsWithUniverseDomain(
+        credentials, builder, endpoint, Credentials.GOOGLE_DEFAULT_UNIVERSE, executorService);
   }
 
   static void registerBuiltinMetricsWithUniverseDomain(
       @Nullable Credentials credentials,
       SdkMeterProviderBuilder builder,
       @Nullable String endpoint,
-      String universeDomain)
+      String universeDomain,
+      @Nullable ScheduledExecutorService executorService)
       throws IOException {
     MetricExporter publicExporter =
         BigtableCloudMonitoringExporter.create(
             "bigtable metrics",
             credentials,
             endpoint,
             universeDomain,
-            new BigtableCloudMonitoringExporter.PublicTimeSeriesConverter());
+            new BigtableCloudMonitoringExporter.PublicTimeSeriesConverter(),
+            executorService);
 
     for (Map.Entry<InstrumentSelector, View> entry :
         BuiltinMetricsConstants.getAllViews().entrySet()) {
       builder.registerView(entry.getKey(), entry.getValue());
     }
-    builder.registerMetricReader(PeriodicMetricReader.create(publicExporter));
+    PeriodicMetricReaderBuilder readerBuilder = PeriodicMetricReader.builder(publicExporter);
+    if (executorService != null) {
+      readerBuilder.setExecutor(executorService);
+    }
+    builder.registerMetricReader(readerBuilder.build());
   }
 }

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

@@ -20,6 +20,7 @@
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
 import java.io.IOException;
+import java.util.concurrent.ScheduledExecutorService;
 
 /**
  * Set a custom OpenTelemetry instance.
@@ -70,26 +71,39 @@ 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 a custom credentials and endpoint. */
+  /** 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 a
+   * shared executor service.
+   */
+  public static void setupSdkMeterProvider(
+      SdkMeterProviderBuilder builder,
+      Credentials credentials,
+      String endpoint,
+      ScheduledExecutorService executor)
+      throws IOException {
+    BuiltinMetricsView.registerBuiltinMetrics(credentials, builder, endpoint, executor);
   }
 
   @Override