fix(bigtable): fall back to classic path when per-RPC CallCredentials are set on session path

The session path (DivertingUnaryCallable) was silently dropping per-RPC
CallCredentials set via GrpcCallContext.withCredentials(), because sessions
are established with channel-level credentials and cannot forward per-RPC
overrides. Any per-RPC credential supplied by the caller was ignored,
causing those calls to use the channel-level credentials instead — a
security bypass.

Fix: detect per-RPC credentials via `context instanceof GrpcCallContext &&
context.getCallOptions().getCredentials() != null` and fall back to the
classic path, which correctly applies the caller-supplied credentials.
The `instanceof` guard is required because calls made without an explicit
ApiCallContext pass a null context (no override → no bypass).

Observability: wire DebugTagTracer through ShimImpl so DivertingUnaryCallable
can record a `per_rpc_credentials_session_fallback` tag into the internal
Cloud Monitoring metric `bigtable.googleapis.com/internal/client/debug_tags`
whenever the fallback fires, giving the server side visibility.

Author: mutianf

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java

@@ -40,6 +40,7 @@
 import com.google.cloud.bigtable.data.v2.internal.compat.ops.ReadRowShimInner;
 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.csm.tracers.DebugTagTracer;
 import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer;
 import com.google.cloud.bigtable.data.v2.internal.dp.DirectAccessInvestigator;
 import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
@@ -81,6 +82,7 @@ public class ShimImpl implements Shim {
   private final ClientConfigurationManager configManager;
   private final Resource<ClientConfigurationManager> configManagerResource;
   private final Client client;
+  private final DebugTagTracer debugTagTracer;
 
   private final ReadRowShimInner readRowShimInner;
   private final MutateRowShim mutateRowShim;
@@ -182,13 +184,20 @@ public static Shim create(
             userCallbackExecutor,
             clientChannelProvider);
 
-    return new ShimImpl(Resource.createOwned(configManager, configManager::close), client);
+    return new ShimImpl(
+        Resource.createOwned(configManager, configManager::close),
+        client,
+        metrics.getDebugTagTracer());
   }
 
-  public ShimImpl(Resource<ClientConfigurationManager> configManagerResource, Client client) {
+  public ShimImpl(
+      Resource<ClientConfigurationManager> configManagerResource,
+      Client client,
+      DebugTagTracer debugTagTracer) {
     this.configManagerResource = configManagerResource;
     this.configManager = configManagerResource.get();
     this.client = client;
+    this.debugTagTracer = debugTagTracer;
 
     this.readRowShimInner = new ReadRowShimInner(client);
     this.mutateRowShim = new MutateRowShim(client);
@@ -221,7 +230,8 @@ public static Shim createForFactoryChild(
             Resource.createShared(userCallbackExecutor),
             Resource.createShared(sharedChannelPool));
 
-    return new ShimImpl(Resource.createShared(sharedConfigManager), client);
+    return new ShimImpl(
+        Resource.createShared(sharedConfigManager), client, metrics.getDebugTagTracer());
   }
 
   /** Returns the raw config manager (e.g. for sharing with factory children). */
@@ -355,13 +365,14 @@ public <RowT> UnaryCallable<Query, RowT> decorateReadRow(
         configManager,
         classic,
         new ReadRowShim<>(readRowShimInner, rowAdapter),
-        Util.extractTimeout(settings));
+        Util.extractTimeout(settings),
+        debugTagTracer);
   }
 
   @Override
   public UnaryCallable<RowMutation, Void> decorateMutateRow(
       UnaryCallable<RowMutation, Void> classic, UnaryCallSettings<?, ?> settings) {
     return new DivertingUnaryCallable<>(
-        configManager, classic, mutateRowShim, Util.extractTimeout(settings));
+        configManager, classic, mutateRowShim, Util.extractTimeout(settings), debugTagTracer);
   }
 }

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ops/DivertingUnaryCallable.java

@@ -22,8 +22,10 @@
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiExceptionFactory;
 import com.google.api.gax.rpc.UnaryCallable;
+import com.google.bigtable.v2.TelemetryConfiguration;
 import com.google.cloud.bigtable.data.v2.internal.compat.FutureAdapter;
 import com.google.cloud.bigtable.data.v2.internal.compat.Util;
+import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DebugTagTracer;
 import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
 import com.google.common.base.Throwables;
 import io.grpc.Context;
@@ -46,23 +48,34 @@ public class DivertingUnaryCallable<ReqT, RespT> extends UnaryCallable<ReqT, Res
   private final UnaryShim<ReqT, RespT> experimental;
 
   private final Duration defaultTimeout;
+  private final DebugTagTracer debugTagTracer;
 
   public DivertingUnaryCallable(
       ClientConfigurationManager configurationManager,
       UnaryCallable<ReqT, RespT> classic,
       UnaryShim<ReqT, RespT> experimental,
-      Duration defaultTimeout) {
+      Duration defaultTimeout,
+      DebugTagTracer debugTagTracer) {
     this.configurationManager = configurationManager;
     this.classic = classic;
     this.experimental = experimental;
     this.defaultTimeout = defaultTimeout;
+    this.debugTagTracer = debugTagTracer;
   }
 
   @Override
   public ApiFuture<RespT> futureCall(ReqT request, ApiCallContext context) {
     if (!useExperimental(request)) {
       return classic.futureCall(request, context);
     }
+    // Per-RPC credential overrides cannot be forwarded to the session path (sessions are
+    // established with channel-level credentials). Fall back to classic to honor the override.
+    if (context instanceof GrpcCallContext
+        && ((GrpcCallContext) context).getCallOptions().getCredentials() != null) {
+      debugTagTracer.record(
+          TelemetryConfiguration.Level.WARN, "per_rpc_credentials_session_fallback");
+      return classic.futureCall(request, context);
+    }
 
     Deadline deadline = Util.extractDeadline((GrpcCallContext) context, defaultTimeout);