feat: add an option to enable DirectPath xDS

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

@@ -265,16 +265,16 @@ private boolean isDirectPathEnabled(String serviceAddress) {
   }
 
   private boolean isDirectPathXdsUsed() {
-    // Method 1: Enable DirectPath xDS by env.
+    // 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;
     }
-    // Method 2: Enable DirectPath xDS by option.
-    if (Boolean.TRUE.equals(useDirectPathXds)) {
-      return true;
-    }
     return false;
   }
 
@@ -704,8 +704,8 @@ public Builder setAllowNonDefaultServiceAccount(boolean allowNonDefaultServiceAc
 
     /** Use DirectPath xDS. Only valid if DirectPath is attempted. */
     @InternalApi("For internal use by google-cloud-java clients only")
-    public Builder setUseDirectPathXds(boolean useDirectPathXds) {
-      this.useDirectPathXds = useDirectPathXds;
+    public Builder setUseDirectPathXds() {
+      this.useDirectPathXds = true;
       return this;
     }
 

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);