fix

Author: sushanb

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessChecker.java

@@ -25,8 +25,10 @@
 import io.grpc.ClientInterceptors;
 import io.grpc.ManagedChannel;
 import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import javax.annotation.Nullable;
 
 /**
  * Evaluates whether a given channel has Direct Access (DirectPath) routing by executing a RPC and
@@ -37,11 +39,15 @@ public class ClassicDirectAccessChecker implements DirectAccessChecker {
   private static final Logger LOG = Logger.getLogger(ClassicDirectAccessChecker.class.getName());
   private final DirectPathCompatibleTracer tracer;
   private final ChannelPrimer channelPrimer;
+  private final ScheduledExecutorService executor;
 
   public ClassicDirectAccessChecker(
-      DirectPathCompatibleTracer tracer, ChannelPrimer channelPrimer) {
+      DirectPathCompatibleTracer tracer,
+      ChannelPrimer channelPrimer,
+      ScheduledExecutorService executor) {
     this.tracer = tracer;
     this.channelPrimer = channelPrimer;
+    this.executor = executor;
   }
 
   @VisibleForTesting
@@ -54,6 +60,7 @@ public boolean check(Channel channel) {
     try {
       return evaluateEligibility(channel);
     } catch (Exception e) {
+      investigateFailure(e);
       LOG.log(Level.WARNING, "Failed to evaluate direct access eligibility.", e);
       return false;
     } finally {
@@ -81,7 +88,16 @@ private boolean evaluateEligibility(Channel channel) {
     if (isEligible) {
       // getIp should be non-null as isEligible is true
       tracer.recordSuccess(sidebandData.getIpProtocol());
+    } else {
+      investigateFailure(null);
     }
     return isEligible;
   }
+
+  @Override
+  public void investigateFailure(@Nullable Throwable originalError) {
+    if (executor != null) {
+      executor.execute(() -> DirectAccessInvestigator.investigateAndReport(tracer, originalError));
+    }
+  }
 }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/DirectAccessChecker.java

@@ -28,4 +28,11 @@ public interface DirectAccessChecker {
    * @return true if the channel is eligible for Direct Access
    */
   boolean check(Channel channel);
+
+  /**
+   * Triggers a investigation into why Direct Access routing failed.
+   *
+   * @param originalError An optional exception that caused the failure.
+   */
+  void investigateFailure(Throwable originalError);
 }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/NoopDirectAccessChecker.java

@@ -0,0 +1,38 @@
+/*
+ * 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.dp;
+
+import com.google.api.core.InternalApi;
+import io.grpc.Channel;
+import javax.annotation.Nullable;
+
+@InternalApi
+public class NoopDirectAccessChecker implements DirectAccessChecker {
+  public static final NoopDirectAccessChecker INSTANCE = new NoopDirectAccessChecker();
+
+  private NoopDirectAccessChecker() {}
+
+  @Override
+  public boolean check(Channel channel) {
+    // If it's disabled, it is never eligible.
+    return false;
+  }
+
+  @Override
+  public void investigateFailure(@Nullable Throwable originalError) {
+    // Do nothing. We don't investigate failures if the feature is disabled.
+  }
+}

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

@@ -33,6 +33,10 @@
 import com.google.cloud.bigtable.data.v2.internal.csm.MetricsImpl;
 import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
 import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer;
+import com.google.cloud.bigtable.data.v2.internal.csm.tracers.NoopDirectPathCompatibleTracer;
+import com.google.cloud.bigtable.data.v2.internal.dp.ClassicDirectAccessChecker;
+import com.google.cloud.bigtable.data.v2.internal.dp.DirectAccessChecker;
+import com.google.cloud.bigtable.data.v2.internal.dp.NoopDirectAccessChecker;
 import com.google.cloud.bigtable.data.v2.stub.metrics.CustomOpenTelemetryMetricsProvider;
 import com.google.cloud.bigtable.gaxx.grpc.BigtableTransportChannelProvider;
 import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer;
@@ -46,7 +50,6 @@
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
-import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -163,18 +166,24 @@ public static BigtableClientContext create(
                 builder.getHeaderProvider().getHeaders());
       }
 
-      Optional<DirectPathCompatibleTracer> optionalTracer =
+      DirectPathCompatibleTracer directPathCompatibleTracer =
           settings.isDirectPathEnabledByDefault()
-              ? Optional.of(metrics.getDirectPathCompatibleTracer())
-              : Optional.empty();
+              ? metrics.getDirectPathCompatibleTracer()
+              : NoopDirectPathCompatibleTracer.INSTANCE;
+
+      DirectAccessChecker directAccessChecker =
+          settings.isDirectPathEnabledByDefault()
+              ? new ClassicDirectAccessChecker(
+                  directPathCompatibleTracer, channelPrimer, backgroundExecutor)
+              : NoopDirectAccessChecker.INSTANCE;
 
       BigtableTransportChannelProvider btTransportProvider =
           BigtableTransportChannelProvider.create(
               transportProvider.build(),
               channelPrimer,
               metrics.getChannelPoolMetricsTracer(),
               backgroundExecutor,
-              optionalTracer);
+              directAccessChecker);
 
       builder.setTransportChannelProvider(btTransportProvider);
     }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java

@@ -23,16 +23,12 @@
 import com.google.api.gax.rpc.TransportChannelProvider;
 import com.google.auth.Credentials;
 import com.google.cloud.bigtable.data.v2.internal.csm.tracers.ChannelPoolMetricsTracer;
-import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer;
-import com.google.cloud.bigtable.data.v2.internal.dp.ClassicDirectAccessChecker;
 import com.google.cloud.bigtable.data.v2.internal.dp.DirectAccessChecker;
-import com.google.cloud.bigtable.data.v2.internal.dp.DirectAccessInvestigator;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStubSettings;
 import com.google.common.base.Preconditions;
 import io.grpc.ManagedChannel;
 import java.io.IOException;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Supplier;
@@ -52,20 +48,19 @@ public final class BigtableTransportChannelProvider implements TransportChannelP
   private final ChannelPrimer channelPrimer;
   @Nullable private final ChannelPoolMetricsTracer channelPoolMetricsTracer;
   @Nullable private final ScheduledExecutorService backgroundExecutor;
-  private final Optional<DirectPathCompatibleTracer> directPathCompatibleTracer;
+  private final DirectAccessChecker directAccessChecker;
 
   private BigtableTransportChannelProvider(
       InstantiatingGrpcChannelProvider instantiatingGrpcChannelProvider,
       ChannelPrimer channelPrimer,
       @Nullable ChannelPoolMetricsTracer channelPoolMetricsTracer,
       @Nullable ScheduledExecutorService backgroundExecutor,
-      Optional<DirectPathCompatibleTracer> directPathCompatibleTracer) {
+      DirectAccessChecker directAccessChecker) {
     delegate = Preconditions.checkNotNull(instantiatingGrpcChannelProvider);
     this.channelPrimer = channelPrimer;
     this.channelPoolMetricsTracer = channelPoolMetricsTracer;
     this.backgroundExecutor = backgroundExecutor;
-    this.directPathCompatibleTracer =
-        directPathCompatibleTracer != null ? directPathCompatibleTracer : Optional.empty();
+    this.directAccessChecker = directAccessChecker;
   }
 
   @Override
@@ -96,7 +91,7 @@ public BigtableTransportChannelProvider withExecutor(Executor executor) {
         channelPrimer,
         channelPoolMetricsTracer,
         backgroundExecutor,
-        directPathCompatibleTracer);
+        directAccessChecker);
   }
 
   @Override
@@ -109,11 +104,7 @@ public TransportChannelProvider withBackgroundExecutor(ScheduledExecutorService
     InstantiatingGrpcChannelProvider newChannelProvider =
         (InstantiatingGrpcChannelProvider) delegate.withBackgroundExecutor(executor);
     return new BigtableTransportChannelProvider(
-        newChannelProvider,
-        channelPrimer,
-        channelPoolMetricsTracer,
-        executor,
-        directPathCompatibleTracer);
+        newChannelProvider, channelPrimer, channelPoolMetricsTracer, executor, directAccessChecker);
   }
 
   @Override
@@ -130,7 +121,7 @@ public BigtableTransportChannelProvider withHeaders(Map<String, String> headers)
         channelPrimer,
         channelPoolMetricsTracer,
         backgroundExecutor,
-        directPathCompatibleTracer);
+        directAccessChecker);
   }
 
   @Override
@@ -147,7 +138,7 @@ public TransportChannelProvider withEndpoint(String endpoint) {
         channelPrimer,
         channelPoolMetricsTracer,
         backgroundExecutor,
-        directPathCompatibleTracer);
+        directAccessChecker);
   }
 
   @Deprecated
@@ -166,7 +157,7 @@ public TransportChannelProvider withPoolSize(int size) {
         channelPrimer,
         channelPoolMetricsTracer,
         backgroundExecutor,
-        directPathCompatibleTracer);
+        directAccessChecker);
   }
 
   /** Expected to only be called once when BigtableClientContext is created */
@@ -179,29 +170,14 @@ public TransportChannel getTransportChannel() throws IOException {
 
     boolean isDirectAccessEligible = false;
 
-    if (!directPathCompatibleTracer.isPresent()) {
-      LOG.fine("Direct access check skipped. Reason: user_disabled or tracer absent");
-    } else {
-      DirectPathCompatibleTracer tracer = directPathCompatibleTracer.get();
-      DirectAccessChecker directAccessChecker =
-          new ClassicDirectAccessChecker(tracer, channelPrimer);
-      try {
-        GrpcTransportChannel grpcTransportChannel =
-            (GrpcTransportChannel) directAccessProvider.getTransportChannel();
-        ManagedChannel directAccessChannel = (ManagedChannel) grpcTransportChannel.getChannel();
-
-        isDirectAccessEligible = directAccessChecker.check(directAccessChannel);
-        if (!isDirectAccessEligible && backgroundExecutor != null) {
-          backgroundExecutor.execute(
-              () -> DirectAccessInvestigator.investigateAndReport(tracer, null));
-        }
-      } catch (Exception e) {
-        LOG.log(Level.FINE, "Client is not direct access eligible, using standard transport.", e);
-        if (backgroundExecutor != null) {
-          backgroundExecutor.execute(
-              () -> DirectAccessInvestigator.investigateAndReport(tracer, e));
-        }
-      }
+    try {
+      GrpcTransportChannel grpcTransportChannel =
+          (GrpcTransportChannel) directAccessProvider.getTransportChannel();
+      ManagedChannel directAccessChannel = (ManagedChannel) grpcTransportChannel.getChannel();
+      isDirectAccessEligible = directAccessChecker.check(directAccessChannel);
+    } catch (Exception e) {
+      LOG.log(Level.WARNING, "Failed to check for direct access.", e);
+      directAccessChecker.investigateFailure(e);
     }
 
     InstantiatingGrpcChannelProvider selectedProvider;
@@ -268,7 +244,7 @@ public TransportChannelProvider withCredentials(Credentials credentials) {
         channelPrimer,
         channelPoolMetricsTracer,
         backgroundExecutor,
-        directPathCompatibleTracer);
+        directAccessChecker);
   }
 
   /** Creates a BigtableTransportChannelProvider. */
@@ -277,12 +253,12 @@ public static BigtableTransportChannelProvider create(
       ChannelPrimer channelPrimer,
       ChannelPoolMetricsTracer outstandingRpcsMetricTracker,
       ScheduledExecutorService backgroundExecutor,
-      Optional<DirectPathCompatibleTracer> directPathCompatibleTracer) {
+      DirectAccessChecker directAccessChecker) {
     return new BigtableTransportChannelProvider(
         instantiatingGrpcChannelProvider,
         channelPrimer,
         outstandingRpcsMetricTracker,
         backgroundExecutor,
-        directPathCompatibleTracer);
+        directAccessChecker);
   }
 }

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessCheckerTest.java

@@ -17,7 +17,13 @@
 
 import static com.google.common.truth.Truth.assertThat;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
 
 import com.google.bigtable.v2.PeerInfo;
 import com.google.cloud.bigtable.data.v2.internal.csm.attributes.Util;
@@ -26,11 +32,13 @@
 import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer;
 import io.grpc.Channel;
 import io.grpc.ManagedChannel;
+import java.util.concurrent.ScheduledExecutorService;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
@@ -45,12 +53,14 @@ public class ClassicDirectAccessCheckerTest {
   @Mock private DirectPathCompatibleTracer mockTracer;
   @Mock private MetadataExtractorInterceptor mockInterceptor;
   @Mock private MetadataExtractorInterceptor.SidebandData mockSidebandData;
+  @Mock private ScheduledExecutorService mockExecutor;
 
   private ClassicDirectAccessChecker checker;
 
   @Before
   public void setUp() throws Exception {
-    checker = spy(new ClassicDirectAccessChecker(mockTracer, mockChannelPrimer));
+    // Pass null for the executor by default so background investigations aren't triggered
+    checker = spy(new ClassicDirectAccessChecker(mockTracer, mockChannelPrimer, null));
     doReturn(mockInterceptor).when(checker).createInterceptor();
     when(mockInterceptor.getSidebandData()).thenReturn(mockSidebandData);
   }
@@ -83,7 +93,7 @@ public void testNotEligibleCFE() {
     boolean isEligible = checker.check(mockChannel);
 
     assertThat(isEligible).isFalse();
-    verifyNoInteractions(mockTracer);
+    verifyNoInteractions(mockTracer); // No interactions because executor is null
     verify(mockChannel).shutdownNow();
   }
 
@@ -122,4 +132,24 @@ public void testNullPeerInfoIsHandledSafely() {
     verifyNoInteractions(mockTracer);
     verify(mockChannel).shutdownNow();
   }
+
+  @Test
+  public void testInvestigationTriggeredOnFailure() {
+    // Re-instantiate the checker with a mock executor to verify investigation is scheduled
+    checker = spy(new ClassicDirectAccessChecker(mockTracer, mockChannelPrimer, mockExecutor));
+    doReturn(mockInterceptor).when(checker).createInterceptor();
+    when(mockInterceptor.getSidebandData()).thenReturn(null); // Force a failure
+
+    boolean isEligible = checker.check(mockChannel);
+
+    assertThat(isEligible).isFalse();
+
+    // Verify the checker submitted a Runnable task to the background executor
+    ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
+    verify(mockExecutor).execute(runnableCaptor.capture());
+
+    // Execute the captured runnable to ensure it safely calls the tracer
+    runnableCaptor.getValue().run();
+    verify(mockTracer).recordFailure(anyString());
+  }
 }