fix(storage): conditionally override authority and add BetaApi annotation

Author: nidhiii-27

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java

@@ -363,17 +363,7 @@ private Tuple<StorageSettings, Opts<UserProject>> resolveSettingsAndOpts() throw
       channelProviderBuilder.setAttemptDirectPathXds();
     }
 
-    if (attemptDirectPathXdsOverInterconnect) {
-      com.google.api.core.ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>
-          existingConfigurator = channelProviderBuilder.getChannelConfigurator();
-      channelProviderBuilder.setChannelConfigurator(
-          channelBuilder -> {
-            if (existingConfigurator != null) {
-              channelBuilder = existingConfigurator.apply(channelBuilder);
-            }
-            return channelBuilder.overrideAuthority(DEFAULT_HOST_NO_SCHEME);
-          });
-    }
+
 
     if (scheme.equals("http")) {
       channelProviderBuilder.setChannelConfigurator(ManagedChannelBuilder::usePlaintext);
@@ -604,6 +594,7 @@ public GrpcStorageOptions.Builder setAttemptDirectPath(boolean attemptDirectPath
      *
      * @since 2.72.0
      */
+    @BetaApi
     public GrpcStorageOptions.Builder setAttemptDirectPathXdsOverInterconnect(
         boolean attemptDirectPathXdsOverInterconnect) {
       this.attemptDirectPathXdsOverInterconnect = attemptDirectPathXdsOverInterconnect;

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageOptionsBuilderTest.java

@@ -92,6 +92,38 @@ public void grpc_attemptDirectPathXdsOverInterconnect() throws Exception {
         (com.google.api.gax.grpc.InstantiatingGrpcChannelProvider) tcp;
 
     assertThat(provider.isAttemptDirectPathXdsOverInterconnect()).isTrue();
+    com.google.api.core.ApiFunction<io.grpc.ManagedChannelBuilder, io.grpc.ManagedChannelBuilder>
+        configurator = provider.toBuilder().getChannelConfigurator();
+    if (configurator != null) {
+      io.grpc.ManagedChannelBuilder<?> fakeBuilder =
+          io.grpc.ManagedChannelBuilder.forAddress("foo", 80);
+      io.grpc.ManagedChannelBuilder<?> appliedBuilder = configurator.apply(fakeBuilder);
+      String authority = extractAuthorityFromChannelBuilder(appliedBuilder);
+      assertThat(authority).isNull();
+    }
+  }
+
+  @Test
+  public void grpc_fallbackFromDirectPathXdsOverInterconnect_doesNotOverrideAuthority()
+      throws Exception {
+    GrpcStorageOptions options =
+        GrpcStorageOptions.grpc()
+            .setHost("https://storage.my-universe.com")
+            .setAttemptDirectPathXdsOverInterconnect(true)
+            .setCredentials(com.google.cloud.NoCredentials.getInstance())
+            .build();
+    com.google.api.gax.grpc.InstantiatingGrpcChannelProvider provider =
+        (com.google.api.gax.grpc.InstantiatingGrpcChannelProvider)
+            options.getStorageSettings().getTransportChannelProvider();
+    com.google.api.core.ApiFunction<io.grpc.ManagedChannelBuilder, io.grpc.ManagedChannelBuilder>
+        configurator = provider.toBuilder().getChannelConfigurator();
+    if (configurator != null) {
+      io.grpc.ManagedChannelBuilder<?> fakeBuilder =
+          io.grpc.ManagedChannelBuilder.forAddress("foo", 80);
+      io.grpc.ManagedChannelBuilder<?> appliedBuilder = configurator.apply(fakeBuilder);
+      String authority = extractAuthorityFromChannelBuilder(appliedBuilder);
+      assertThat(authority).isNull();
+    }
   }
 
   @Test
@@ -115,4 +147,43 @@ public ResultRetryAlgorithm<?> getNonidempotentHandler() {
       return null;
     }
   }
+
+  private static String extractAuthorityFromChannelBuilder(
+      io.grpc.ManagedChannelBuilder<?> channelBuilder) {
+    try {
+      Object current = channelBuilder;
+      while (current != null) {
+        Class<?> clazz = current.getClass();
+        java.lang.reflect.Field field = null;
+        while (clazz != null) {
+          try {
+            field = clazz.getDeclaredField("authority");
+            break;
+          } catch (NoSuchFieldException e) {
+            try {
+              field = clazz.getDeclaredField("overrideAuthority");
+              break;
+            } catch (Exception ignored) {
+            }
+            clazz = clazz.getSuperclass();
+          }
+        }
+        if (field != null) {
+          field.setAccessible(true);
+          return (String) field.get(current);
+        }
+        try {
+          java.lang.reflect.Field delegate = current.getClass().getDeclaredField("delegate");
+          delegate.setAccessible(true);
+          current = delegate.get(current);
+        } catch (Exception e) {
+          break;
+        }
+      }
+      return null;
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
 }
+

sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java

@@ -771,6 +771,9 @@ public ManagedChannelBuilder<?> createChannelBuilder() throws IOException {
             target += "?force-xds";
           }
           builder = Grpc.newChannelBuilder(target, channelCreds);
+          if (isAttemptDirectPathXdsOverInterconnect() && serviceAddress.equals("storage-direct.googleapis.com")) {
+            builder.overrideAuthority("storage.googleapis.com");
+          }
           resolvedTarget = target;
         } else {
           builder = Grpc.newChannelBuilderForAddress(serviceAddress, port, channelCreds);
@@ -787,6 +790,18 @@ public ManagedChannelBuilder<?> createChannelBuilder() throws IOException {
               Level.WARNING,
               "DirectPath was requested but is not available. Falling back to CloudPath.");
         }
+        String fallbackEndpoint = endpoint;
+        String fallbackMtlsEndpoint = mtlsEndpoint;
+        if (serviceAddress.equals("storage-direct.googleapis.com")) {
+          serviceAddress = "storage.googleapis.com";
+          fallbackEndpoint = serviceAddress + ":" + port;
+          if (fallbackMtlsEndpoint != null
+              && fallbackMtlsEndpoint.contains("storage-direct.googleapis.com")) {
+            fallbackMtlsEndpoint =
+                fallbackMtlsEndpoint.replace(
+                    "storage-direct.googleapis.com", "storage.googleapis.com");
+          }
+        }
         ChannelCredentials channelCredentials;
         try {
           // Try and create credentials via DCA. See https://google.aip.dev/auth/4114.
@@ -796,8 +811,8 @@ public ManagedChannelBuilder<?> createChannelBuilder() throws IOException {
         }
         if (channelCredentials != null) {
           // Create the channel using channel credentials created via DCA.
-          builder = Grpc.newChannelBuilder(endpoint, channelCredentials);
-          resolvedTarget = endpoint;
+          builder = Grpc.newChannelBuilder(fallbackEndpoint, channelCredentials);
+          resolvedTarget = fallbackEndpoint;
         } else {
           // Could not create channel credentials via DCA. In accordance with
           // https://google.aip.dev/auth/4115, if credentials not available through
@@ -815,11 +830,12 @@ public ManagedChannelBuilder<?> createChannelBuilder() throws IOException {
             }
             // Connect to the MTLS endpoint when using S2A because S2A is used to perform an MTLS
             // handshake.
-            builder = Grpc.newChannelBuilder(mtlsEndpoint, channelCredentials);
-            resolvedTarget = mtlsEndpoint;
+            builder = Grpc.newChannelBuilder(fallbackMtlsEndpoint, channelCredentials);
+            resolvedTarget = fallbackMtlsEndpoint;
           } else {
             // Use default if we cannot initialize channel credentials via DCA or S2A.
             builder = ManagedChannelBuilder.forAddress(serviceAddress, port);
+            builder.overrideAuthority(serviceAddress);
             resolvedTarget = serviceAddress + ":" + port;
           }
         }

sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java

@@ -1786,6 +1786,8 @@ void canUseDirectPath_interconnectEnabledButDirectPathDisabled_fallsBackToCloudP
     assertThat(logHandler.getAllMessages())
         .contains("DirectPath was requested but is not available. Falling back to CloudPath.");
     InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler);
+    String authority = extractAuthorityFromChannelBuilder(provider.createChannelBuilder());
+    Truth.assertThat(authority).isNotEqualTo("storage.googleapis.com");
   }
 
   @Test
@@ -1818,6 +1820,8 @@ void canUseDirectPath_interconnectAndDirectPathEnabledButNonGduUniverse_fallsBac
     assertThat(logHandler.getAllMessages())
         .contains("DirectPath was requested but is not available. Falling back to CloudPath.");
     InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler);
+    String authority = extractAuthorityFromChannelBuilder(provider.createChannelBuilder());
+    Truth.assertThat(authority).isNotEqualTo("storage.googleapis.com");
   }
 
   private static class FakeLogHandler extends Handler {
@@ -1839,4 +1843,43 @@ List<String> getAllMessages() {
       return records.stream().map(LogRecord::getMessage).collect(Collectors.toList());
     }
   }
+
+
+  private static String extractAuthorityFromChannelBuilder(
+      ManagedChannelBuilder<?> channelBuilder) {
+    try {
+      Object current = channelBuilder;
+      while (current != null) {
+        Class<?> clazz = current.getClass();
+        java.lang.reflect.Field field = null;
+        while (clazz != null) {
+          try {
+            field = clazz.getDeclaredField("authority");
+            break;
+          } catch (NoSuchFieldException e) {
+            try {
+              field = clazz.getDeclaredField("overrideAuthority");
+              break;
+            } catch (Exception ignored) {
+            }
+            clazz = clazz.getSuperclass();
+          }
+        }
+        if (field != null) {
+          field.setAccessible(true);
+          return (String) field.get(current);
+        }
+        try {
+          java.lang.reflect.Field delegate = current.getClass().getDeclaredField("delegate");
+          delegate.setAccessible(true);
+          current = delegate.get(current);
+        } catch (Exception e) {
+          break;
+        }
+      }
+      return null;
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
 }