feat: add an option to enable DirectPath xDS

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

@@ -109,6 +109,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   @Nullable private final Credentials credentials;
   @Nullable private final ChannelPrimer channelPrimer;
   @Nullable private final Boolean attemptDirectPath;
+  @Nullable private final Boolean useDirectPathXds;
   @Nullable private final Boolean allowNonDefaultServiceAccount;
   @VisibleForTesting final ImmutableMap<String, ?> directPathServiceConfig;
   @Nullable private final MtlsProvider mtlsProvider;
@@ -134,6 +135,7 @@ private InstantiatingGrpcChannelProvider(Builder builder) {
     this.credentials = builder.credentials;
     this.channelPrimer = builder.channelPrimer;
     this.attemptDirectPath = builder.attemptDirectPath;
+    this.useDirectPathXds = builder.useDirectPathXds;
     this.allowNonDefaultServiceAccount = builder.allowNonDefaultServiceAccount;
     this.directPathServiceConfig =
         builder.directPathServiceConfig == null
@@ -262,6 +264,20 @@ private boolean isDirectPathEnabled(String serviceAddress) {
     return false;
   }
 
+  private boolean isDirectPathXdsUsed() {
+    // Method 1: Enable DirectPath xDS by option.
+    if (Boolean.TRUE.equals(useDirectPathXds)) {
+      return true;
+    }
+    // Method 2: Enable DirectPath xDS by env.
+    String directPathXdsEnv = envProvider.getenv(DIRECT_PATH_ENV_ENABLE_XDS);
+    boolean isDirectPathXdsEnv = Boolean.parseBoolean(directPathXdsEnv);
+    if (isDirectPathXdsEnv) {
+      return true;
+    }
+    return false;
+  }
+
   private boolean isNonDefaultServiceAccountAllowed() {
     if (allowNonDefaultServiceAccount != null && allowNonDefaultServiceAccount) {
       return true;
@@ -324,7 +340,7 @@ && isOnComputeEngine()) {
       CallCredentials callCreds = MoreCallCredentials.from(credentials);
       ChannelCredentials channelCreds =
           GoogleDefaultChannelCredentials.newBuilder().callCredentials(callCreds).build();
-      isDirectPathXdsEnabled = Boolean.parseBoolean(envProvider.getenv(DIRECT_PATH_ENV_ENABLE_XDS));
+      isDirectPathXdsEnabled = isDirectPathXdsUsed();
       if (isDirectPathXdsEnabled) {
         // google-c2p: CloudToProd(C2P) Directpath. This scheme is defined in
         // io.grpc.googleapis.GoogleCloudToProdNameResolverProvider.
@@ -450,6 +466,7 @@ public static final class Builder {
     @Nullable private ChannelPrimer channelPrimer;
     private ChannelPoolSettings channelPoolSettings;
     @Nullable private Boolean attemptDirectPath;
+    @Nullable private Boolean useDirectPathXds;
     @Nullable private Boolean allowNonDefaultServiceAccount;
     @Nullable private ImmutableMap<String, ?> directPathServiceConfig;
 
@@ -476,6 +493,7 @@ private Builder(InstantiatingGrpcChannelProvider provider) {
       this.channelPrimer = provider.channelPrimer;
       this.channelPoolSettings = provider.channelPoolSettings;
       this.attemptDirectPath = provider.attemptDirectPath;
+      this.useDirectPathXds = provider.useDirectPathXds;
       this.allowNonDefaultServiceAccount = provider.allowNonDefaultServiceAccount;
       this.directPathServiceConfig = provider.directPathServiceConfig;
       this.mtlsProvider = provider.mtlsProvider;
@@ -684,6 +702,13 @@ public Builder setAllowNonDefaultServiceAccount(boolean allowNonDefaultServiceAc
       return this;
     }
 
+    /** Use DirectPath xDS. Only valid if DirectPath is attempted. */
+    @InternalApi("For internal use by google-cloud-java clients only")
+    public Builder setUseDirectPathXds() {
+      this.useDirectPathXds = true;
+      return this;
+    }
+
     /**
      * Sets a service config for direct path. If direct path is not enabled, the provided service
      * config will be ignored.

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

@@ -273,6 +273,31 @@ public void testWithGCECredentials() throws IOException {
     provider.getTransportChannel().shutdownNow();
   }
 
+  @Test
+  public void testDirectPathXds() throws IOException {
+    ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
+    executor.shutdown();
+
+    TransportChannelProvider provider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setAttemptDirectPath(true)
+            .setUseDirectPathXds()
+            .build()
+            .withExecutor((Executor) executor)
+            .withHeaders(Collections.<String, String>emptyMap())
+            .withEndpoint("localhost:8080");
+
+    assertThat(provider.needsCredentials()).isTrue();
+    if (InstantiatingGrpcChannelProvider.isOnComputeEngine()) {
+      provider = provider.withCredentials(ComputeEngineCredentials.create());
+    } else {
+      provider = provider.withCredentials(CloudShellCredentials.create(3000));
+    }
+    assertThat(provider.needsCredentials()).isFalse();
+
+    provider.getTransportChannel().shutdownNow();
+  }
+
   @Test
   public void testWithNonGCECredentials() throws IOException {
     ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);