feat: add new session based protocol stack (#2862)

The new protocol changes the paradigm of bigtable service from an RPC server where each operation is independent to more of a file like model where a table is opened and allows many read & write operations. This will have a significant impact on lowering latencies. All of the changes are internal to the client and the existing public surface remains the same. By default the new protocol is disabled and will be slowly enabled in the future using the client config api.

Notable changes:
- When the new protocol is enabled, the BigtableClientFactory doesnt share the underlying ChannelPool. The new ChannelPool is tracks the connected server, so sharing a channel between multiple resource is no longer possible. 
- The new protocol does not support custom TransportProviders, setting one will force a fallback to the classic protocol
- Retries have been revamped: 
   - the client now tracks how far an rpc went which allows the client to retry non-idempotent rpcs safely
   - attempt timeouts are no longer relevant, they have been replaced with heartbeats
   - Mutations that are not idempotent (have serverside timestamps or counters) are now correctly identified and will be retried if they left the client
- The new transport supports the ambient grpc Context and will consistently respect the ambient deadlines and cancellation
- Customers can opt out of the slow rollout by setting the env var `CBT_DISABLE_SESSIONS=true`

Author: igorbernstein2

generation_config.yaml

@@ -1,5 +1,5 @@
 gapic_generator_version: 2.68.0
-googleapis_commitish: cd090841ab172574e740c214c99df00aef9c0dee
+googleapis_commitish: f5cb7afc40b63d52f43bc306cb9b64a87b681aea
 libraries_bom_version: 26.79.0
 template_excludes:
   - .gitignore

google-cloud-bigtable/pom.xml

@@ -159,15 +159,9 @@
       <artifactId>google-http-client-gson</artifactId>
       <scope>runtime</scope>
     </dependency>
-    <!--
-      grpc-stub is needed directly by our tests and transitively by grpc-alts at runtime.
-      So it has to be declared as a direct dependency and to avoid overriding grpc-alts'
-      runtime requirement it has to be promoted to the runtime scope.
-    -->
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-stub</artifactId>
-      <scope>runtime</scope>
     </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
@@ -225,6 +219,10 @@
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-context</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-sdk</artifactId>

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

@@ -74,6 +74,10 @@ public final class BigtableDataClientFactory implements AutoCloseable {
    */
   public static BigtableDataClientFactory create(BigtableDataSettings defaultSettings)
       throws IOException {
+    BigtableDataSettings.Builder builder = defaultSettings.toBuilder();
+    builder.stubSettings().setSessionsEnabled(false);
+    defaultSettings = builder.build();
+
     BigtableClientContext sharedClientContext =
         BigtableClientContext.create(defaultSettings.getStubSettings());
     ClientOperationSettings perOpSettings = defaultSettings.getStubSettings().getPerOpSettings();

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

@@ -133,6 +133,7 @@ public static Builder newBuilderForEmulator(String hostname, int port) {
             .setMetricsProvider(
                 NoopMetricsProvider.INSTANCE) // disable exporting metrics for emulator
             .disableInternalMetrics()
+            .setSessionsEnabled(false)
             .setTransportChannelProvider(
                 InstantiatingGrpcChannelProvider.newBuilder()
                     .setMaxInboundMessageSize(256 * 1024 * 1024)

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2026 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.internal.api;
+
+import com.google.bigtable.v2.FeatureFlags;
+import com.google.bigtable.v2.OpenAuthorizedViewRequest;
+import com.google.bigtable.v2.OpenAuthorizedViewRequest.Permission;
+import com.google.bigtable.v2.SessionMutateRowRequest;
+import com.google.bigtable.v2.SessionMutateRowResponse;
+import com.google.bigtable.v2.SessionReadRowRequest;
+import com.google.bigtable.v2.SessionReadRowResponse;
+import com.google.cloud.bigtable.data.v2.internal.channels.ChannelPool;
+import com.google.cloud.bigtable.data.v2.internal.csm.Metrics;
+import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
+import com.google.cloud.bigtable.data.v2.internal.session.SessionPool;
+import com.google.cloud.bigtable.data.v2.internal.session.VRpcDescriptor;
+import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
+import io.grpc.CallOptions;
+import io.grpc.Deadline;
+import java.io.Closeable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+
+public class AuthorizedViewAsync implements AutoCloseable, Closeable {
+
+  private final TableBase base;
+
+  static AuthorizedViewAsync createAndStart(
+      FeatureFlags featureFlags,
+      ClientInfo clientInfo,
+      ClientConfigurationManager configManager,
+      ChannelPool channelPool,
+      CallOptions callOptions,
+      String tableId,
+      String viewId,
+      Permission permission,
+      Metrics metrics,
+      ScheduledExecutorService executorService) {
+
+    AuthorizedViewName viewName =
+        AuthorizedViewName.builder()
+            .setProjectId(clientInfo.getInstanceName().getProjectId())
+            .setInstanceId(clientInfo.getInstanceName().getInstanceId())
+            .setTableId(tableId)
+            .setAuthorizedViewId(viewId)
+            .build();
+
+    OpenAuthorizedViewRequest openRequest =
+        OpenAuthorizedViewRequest.newBuilder()
+            .setAuthorizedViewName(viewName.toString())
+            .setAppProfileId(clientInfo.getAppProfileId())
+            .setPermission(permission)
+            .build();
+
+    TableBase base =
+        TableBase.createAndStart(
+            openRequest,
+            VRpcDescriptor.AUTHORIZED_VIEW_SESSION,
+            VRpcDescriptor.READ_ROW_AUTH_VIEW,
+            VRpcDescriptor.MUTATE_ROW_AUTH_VIEW,
+            featureFlags,
+            clientInfo,
+            configManager,
+            channelPool,
+            callOptions,
+            viewName.toString(),
+            metrics,
+            executorService);
+
+    return new AuthorizedViewAsync(base);
+  }
+
+  AuthorizedViewAsync(TableBase viewBase) {
+    this.base = viewBase;
+  }
+
+  public SessionPool<?> getSessionPool() {
+    return base.getSessionPool();
+  }
+
+  public CompletableFuture<SessionReadRowResponse> readRow(
+      SessionReadRowRequest req, Deadline deadline) {
+    UnaryResponseFuture<SessionReadRowResponse> f = new UnaryResponseFuture<>();
+    base.readRow(req, f, deadline);
+    return f;
+  }
+
+  public CompletableFuture<SessionMutateRowResponse> mutateRow(
+      SessionMutateRowRequest req, Deadline deadline) {
+    UnaryResponseFuture<SessionMutateRowResponse> f = new UnaryResponseFuture<>();
+    base.mutateRow(req, f, deadline);
+    return f;
+  }
+
+  @Override
+  public void close() {
+    this.base.close();
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewName.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2026 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.internal.api;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Splitter;
+import java.util.List;
+
+@AutoValue
+public abstract class AuthorizedViewName {
+
+  public abstract String getProjectId();
+
+  public abstract String getInstanceId();
+
+  public abstract String getTableId();
+
+  public abstract String getAuthorizedViewId();
+
+  public InstanceName getInstanceName() {
+    return InstanceName.builder()
+        .setProjectId(getProjectId())
+        .setInstanceId(getInstanceId())
+        .build();
+  }
+
+  public TableName getTableName() {
+    return TableName.builder()
+        .setProjectId(getProjectId())
+        .setInstanceId(getInstanceId())
+        .setTableId(getTableId())
+        .build();
+  }
+
+  @Override
+  public final String toString() {
+    return String.format("%s/authorizedViews/%s", getTableName(), getAuthorizedViewId());
+  }
+
+  public static AuthorizedViewName of(
+      String projectId, String instanceId, String tableId, String viewId) {
+    return builder()
+        .setProjectId(projectId)
+        .setInstanceId(instanceId)
+        .setTableId(tableId)
+        .setAuthorizedViewId(viewId)
+        .build();
+  }
+
+  public static Builder builder() {
+    return new AutoValue_AuthorizedViewName.Builder();
+  }
+
+  public static AuthorizedViewName parse(String name) {
+    List<String> parts = Splitter.on('/').splitToList(name);
+    Preconditions.checkArgument(parts.size() == 8, "Invalid authorized view name: %s", name);
+    Preconditions.checkArgument(
+        "projects".equals(parts.get(0)),
+        "Invalid authorized view name: %s, must start with projects/",
+        name);
+    Preconditions.checkArgument(
+        !parts.get(1).isEmpty(), "Invalid authorized view name %s, must have a project id", name);
+    Preconditions.checkArgument(
+        "instances".equals(parts.get(2)),
+        "Invalid authorized view name: %s, must start with projects/$PROJECT_ID/instances/",
+        name);
+    Preconditions.checkArgument(
+        !parts.get(3).isEmpty(), "Invalid authorized view name %s, must have an instance id", name);
+    Preconditions.checkArgument(
+        "tables".equals(parts.get(4)),
+        "Invalid authorized view name: %s, must start with projects/$PROJECT_ID/instances/$INSTANCE_ID/tables",
+        name);
+    Preconditions.checkArgument(
+        !parts.get(5).isEmpty(), "Invalid authorized view name %s, must have table id", name);
+    Preconditions.checkArgument(
+        "authorizedViews".equals(parts.get(6)),
+        "Invalid authorized view name: %s, must start with projects/$PROJECT_ID/instances/$INSTANCE_ID/tables/$TABLE_ID/authorizedViews",
+        name);
+    Preconditions.checkArgument(
+        !parts.get(7).isEmpty(),
+        "Invalid authorized view name %s, must have authorized view id",
+        name);
+
+    return builder()
+        .setProjectId(parts.get(1))
+        .setInstanceId(parts.get(3))
+        .setTableId(parts.get(5))
+        .setAuthorizedViewId(parts.get(7))
+        .build();
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder setProjectId(String projectId);
+
+    public abstract Builder setInstanceId(String instanceId);
+
+    public abstract Builder setTableId(String tableId);
+
+    public abstract Builder setAuthorizedViewId(String viewId);
+
+    public abstract AuthorizedViewName build();
+  }
+}