Merge branch 'main' into view

Author: mutianf

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

@@ -34,6 +34,13 @@
         <className>com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub</className>
         <method>*</method>
     </difference>
+    <difference>
+        <!-- change method types is ok because EnhancedBigtableStub is InternalApi -->
+        <differenceType>7005</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub</className>
+        <method>*</method>
+        <to>*</to>
+    </difference>
     <difference>
         <!-- method name change is ok because EnhancedBigtableStub is InternalApi -->
         <differenceType>7002</differenceType>
@@ -504,4 +511,16 @@
         <className>com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsConstants</className>
         <method>*</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/BigtableDataClient.java

@@ -25,7 +25,6 @@
 import com.google.api.gax.batching.Batcher;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.rpc.ApiExceptions;
-import com.google.api.gax.rpc.ClientContext;
 import com.google.api.gax.rpc.ResponseObserver;
 import com.google.api.gax.rpc.ServerStream;
 import com.google.api.gax.rpc.ServerStreamingCallable;
@@ -55,6 +54,7 @@
 import com.google.cloud.bigtable.data.v2.models.sql.PreparedStatement;
 import com.google.cloud.bigtable.data.v2.models.sql.ResultSet;
 import com.google.cloud.bigtable.data.v2.models.sql.SqlType;
+import com.google.cloud.bigtable.data.v2.stub.BigtableClientContext;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStub;
 import com.google.cloud.bigtable.data.v2.stub.sql.SqlServerStream;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -186,7 +186,7 @@ public static BigtableDataClient create(BigtableDataSettings settings) throws IO
    * BigtableDataClientFactory#close()} is called.
    */
   static BigtableDataClient createWithClientContext(
-      BigtableDataSettings settings, ClientContext context) throws IOException {
+      BigtableDataSettings settings, BigtableClientContext context) throws IOException {
     EnhancedBigtableStub stub =
         EnhancedBigtableStub.createWithClientContext(settings.getStubSettings(), context);
     return new BigtableDataClient(stub);

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

@@ -116,7 +116,8 @@ public BigtableDataClient createDefault() {
                       sharedClientContext.getInternalOpenTelemtry()))
               .build();
 
-      return BigtableDataClient.createWithClientContext(defaultSettings, clientContext);
+      return BigtableDataClient.createWithClientContext(
+          defaultSettings, sharedClientContext.withClientContext(clientContext));
     } catch (IOException e) {
       // Should never happen because the connection has been established already
       throw new RuntimeException(
@@ -145,7 +146,8 @@ public BigtableDataClient createForAppProfile(@Nonnull String appProfileId) thro
                     sharedClientContext.getOpenTelemetry(),
                     sharedClientContext.getInternalOpenTelemtry()))
             .build();
-    return BigtableDataClient.createWithClientContext(settings, clientContext);
+    return BigtableDataClient.createWithClientContext(
+        settings, sharedClientContext.withClientContext(clientContext));
   }
 
   /**
@@ -175,7 +177,8 @@ public BigtableDataClient createForInstance(@Nonnull String projectId, @Nonnull
                     sharedClientContext.getInternalOpenTelemtry()))
             .build();
 
-    return BigtableDataClient.createWithClientContext(settings, clientContext);
+    return BigtableDataClient.createWithClientContext(
+        settings, sharedClientContext.withClientContext(clientContext));
   }
 
   /**
@@ -204,6 +207,7 @@ public BigtableDataClient createForInstance(
                     sharedClientContext.getOpenTelemetry(),
                     sharedClientContext.getInternalOpenTelemtry()))
             .build();
-    return BigtableDataClient.createWithClientContext(settings, clientContext);
+    return BigtableDataClient.createWithClientContext(
+        settings, sharedClientContext.withClientContext(clientContext));
   }
 }

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() {
@@ -201,6 +223,15 @@ public ClientContext getClientContext() {
     return this.clientContext;
   }
 
+  public BigtableClientContext withClientContext(ClientContext clientContext) {
+    return new BigtableClientContext(
+        clientContext,
+        openTelemetry,
+        internalOpenTelemetry,
+        metricsProvider,
+        backgroundExecutorProvider);
+  }
+
   public void close() throws Exception {
     for (BackgroundResource resource : clientContext.getBackgroundResources()) {
       resource.close();
@@ -211,13 +242,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 =
@@ -229,7 +264,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;
+  }
+}