fix: fallback on VPC (#13567)

Author: mutianf

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

@@ -27,6 +27,7 @@
 import io.grpc.ManagedChannel;
 import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
+import java.util.Locale;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Level;
@@ -74,6 +75,15 @@ public boolean check(Channel channel) {
     }
   }
 
+  /** Checks if the exception is due to a VPC Service Controls policy violation. */
+  private boolean isVpcServiceControlsViolation(StatusRuntimeException e) {
+    String description = e.getStatus().getDescription();
+    String message = e.getMessage();
+    String expected = "request is prohibited by organization's policy";
+    return (description != null && description.toLowerCase(Locale.ROOT).contains(expected))
+        || (message != null && message.toLowerCase(Locale.ROOT).contains(expected));
+  }
+
   /** Executes the underlying RPC and evaluates the eligibility. */
   private boolean evaluateEligibility(Channel channel) {
     MetadataExtractorInterceptor interceptor = createInterceptor();
@@ -91,8 +101,13 @@ private boolean evaluateEligibility(Channel channel) {
       if (e.getStatus().getCode() != Code.PERMISSION_DENIED) {
         throw e;
       }
-      // Failed with permission error, resorting to ALTS check.
-      isEligible = sidebandData.isAlts();
+
+      if (isVpcServiceControlsViolation(e)) {
+        LOG.log(Level.WARNING, "DirectPath is blocked by a perimeter policy violation.");
+      } else {
+        // Failed with standard permission error, resorting to ALTS check.
+        isEligible = sidebandData.isAlts();
+      }
     }
 
     if (isEligible) {