[Java] Honor GOOGLE_CLOUD_UNIVERSE_DOMAIN for the Dataflow endpoint

CHANGES.md

@@ -90,6 +90,7 @@
 * (Java) JmsIO migrated to `jakarta.jms` (JMS 3.1) and ActiveMQ 6.2.5. User code implementing `JmsIO.MessageMapper`,
   `valueMapper`, `topicNameMapper`, or providing a `ConnectionFactory` must update imports from `javax.jms.*` to
   `jakarta.jms.*`. The module now requires Java 17 at runtime ([#38729](https://github.com/apache/beam/issues/38729)).
+* (Java) Dataflow runner: `--apiRootUrl` no longer defaults to `https://dataflow.googleapis.com/`. When neither `--apiRootUrl` nor an absolute `--dataflowEndpoint` is set, the Google API client library resolves the Dataflow endpoint and honors `GOOGLE_CLOUD_UNIVERSE_DOMAIN`, so Dataflow jobs can be submitted in non-default universes. Code that reads `DataflowPipelineDebugOptions.getApiRootUrl()` should handle a `null` value ([#40569](https://github.com/apache/beam/issues/40569)).
 * (Python) `Regex.find_all` with `group=Regex.ALL` now returns every group, as `(match, group1, group2, ...)`, instead of only group 1, and with `outputEmpty=False` it drops a match only when the whole match is empty, matching the `group=0` behavior ([#40399](https://github.com/apache/beam/pull/40399)).
 * (Python) `Regex.all_matches` now returns every group of the regex, instead of failing with a `TypeError` when the regex has no groups and dropping nested or unmatched trailing groups ([#40562](https://github.com/apache/beam/issues/40562)).
 * (Python) `--dataflow_endpoint` now defaults to `''` instead of `https://dataflow.googleapis.com`. When unset, the Google Cloud client library resolves the Dataflow endpoint and honors `GOOGLE_CLOUD_UNIVERSE_DOMAIN`, so Dataflow jobs can be submitted in non-default universes. Code that reads `GoogleCloudOptions.dataflow_endpoint` should handle an empty value ([#40569](https://github.com/apache/beam/issues/40569)).

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java

@@ -31,6 +31,7 @@
 import org.apache.beam.sdk.options.MemoryMonitorOptions;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.util.InstanceBuilder;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * Internal. Options used to control execution of the Dataflow SDK for debugging and testing
@@ -47,15 +48,22 @@ public interface DataflowPipelineDebugOptions
    * The root URL for the Dataflow API. {@code dataflowEndpoint} can override this value if it
    * contains an absolute URL, otherwise {@code apiRootUrl} will be combined with {@code
    * dataflowEndpoint} to generate the full URL to communicate with the Dataflow API.
+   *
+   * <p>If not set, the root URL is resolved by the Google API client library, which honors {@code
+   * GOOGLE_CLOUD_UNIVERSE_DOMAIN} and otherwise uses the default public URL ({@value
+   * Dataflow#DEFAULT_ROOT_URL}).
    */
   @Description(
       "The root URL for the Dataflow API. dataflowEndpoint can override this "
           + "value if it contains an absolute URL, otherwise apiRootUrl will be combined with "
-          + "dataflowEndpoint to generate the full URL to communicate with the Dataflow API.")
-  @Default.String(Dataflow.DEFAULT_ROOT_URL)
-  String getApiRootUrl();
+          + "dataflowEndpoint to generate the full URL to communicate with the Dataflow API. "
+          + "If not set, the root URL is resolved by the Google API client library, which "
+          + "honors GOOGLE_CLOUD_UNIVERSE_DOMAIN and otherwise uses the default public URL ("
+          + Dataflow.DEFAULT_ROOT_URL
+          + ").")
+  @Nullable String getApiRootUrl();
 
-  void setApiRootUrl(String value);
+  void setApiRootUrl(@Nullable String value);
 
   /**
    * Dataflow endpoint to use.
@@ -64,7 +72,8 @@ public interface DataflowPipelineDebugOptions
    * SDK version was released.
    *
    * <p>If the string contains "://", then this is treated as a URL, otherwise {@link
-   * #getApiRootUrl()} is used as the root URL.
+   * #getApiRootUrl()} is used as the root URL (or, if that is not set, the root URL resolved by the
+   * Google API client library).
    */
   @Description(
       "The URL for the Dataflow API. If the string contains \"://\", this"

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java

@@ -161,7 +161,8 @@ public interface DataflowPipelineOptions
    * SDK version was released.
    *
    * <p>If the string contains "://", then this is treated as a URL, otherwise {@link
-   * #getApiRootUrl()} is used as the root URL.
+   * #getApiRootUrl()} is used as the root URL (or, if that is not set, the root URL resolved by the
+   * Google API client library).
    */
   @Description(
       "The URL for the Dataflow API. If the string contains \"://\", this"

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTransport.java

@@ -30,7 +30,9 @@
 import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
 import org.apache.beam.sdk.extensions.gcp.auth.NullCredentialInitializer;
 import org.apache.beam.sdk.extensions.gcp.util.RetryHttpRequestInitializer;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /** Helpers for cloud communication. */
 public class DataflowTransport {
@@ -59,28 +61,42 @@ private static ApiComponents apiComponentsFromUrl(String urlString) {
     }
   }
 
-  /** Returns a Google Cloud Dataflow client builder. */
+  /**
+   * Returns a Google Cloud Dataflow client builder.
+   *
+   * <p>The root URL and service path are only set on the builder when the user configured them via
+   * {@code dataflowEndpoint} and/or {@code apiRootUrl}. Otherwise, the endpoint is resolved by the
+   * Google API client library, which honors {@code GOOGLE_CLOUD_UNIVERSE_DOMAIN} and otherwise uses
+   * the default public URL. Note that calling {@link Dataflow.Builder#setRootUrl} (even with the
+   * default URL) disables this resolution.
+   */
   public static Dataflow.Builder newDataflowClient(DataflowPipelineOptions options) {
+    Dataflow.Builder builder =
+        new Dataflow.Builder(
+                getTransport(),
+                getJsonFactory(),
+                chainHttpRequestInitializer(
+                    options.getGcpCredential(),
+                    // Do not log 404. It clutters the output and is possibly even required by the
+                    // caller.
+                    new RetryHttpRequestInitializer(ImmutableList.of(404))))
+            .setApplicationName(options.getAppName())
+            .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
+
     String servicePath = options.getDataflowEndpoint();
-    ApiComponents components;
-    if (servicePath.contains("://")) {
-      components = apiComponentsFromUrl(servicePath);
+    @Nullable String apiRootUrl = options.getApiRootUrl();
+    if (!Strings.isNullOrEmpty(servicePath) && servicePath.contains("://")) {
+      ApiComponents components = apiComponentsFromUrl(servicePath);
+      builder.setRootUrl(components.rootUrl).setServicePath(components.servicePath);
     } else {
-      components = new ApiComponents(options.getApiRootUrl(), servicePath);
+      if (apiRootUrl != null && !apiRootUrl.isEmpty()) {
+        builder.setRootUrl(apiRootUrl);
+      }
+      if (!Strings.isNullOrEmpty(servicePath)) {
+        builder.setServicePath(servicePath);
+      }
     }
-
-    return new Dataflow.Builder(
-            getTransport(),
-            getJsonFactory(),
-            chainHttpRequestInitializer(
-                options.getGcpCredential(),
-                // Do not log 404. It clutters the output and is possibly even required by the
-                // caller.
-                new RetryHttpRequestInitializer(ImmutableList.of(404))))
-        .setApplicationName(options.getAppName())
-        .setRootUrl(components.rootUrl)
-        .setServicePath(components.servicePath)
-        .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
+    return builder;
   }
 
   private static HttpRequestInitializer chainHttpRequestInitializer(

runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/DataflowTransportTest.java

@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://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 org.apache.beam.runners.dataflow.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assume.assumeTrue;
+
+import com.google.api.services.dataflow.Dataflow;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
+import org.apache.beam.sdk.extensions.gcp.auth.TestCredential;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link DataflowTransport}. */
+@RunWith(JUnit4.class)
+public class DataflowTransportTest {
+
+  private static final String UNIVERSE_DOMAIN = "example-universe.goog";
+
+  private DataflowPipelineOptions options;
+
+  @Before
+  public void setUp() {
+    options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
+    options.setGcpCredential(new TestCredential());
+  }
+
+  @Test
+  public void testApiRootUrlIsUnsetByDefault() {
+    assertNull(options.getApiRootUrl());
+    assertEquals("", options.getDataflowEndpoint());
+  }
+
+  @Test
+  public void testDefaultEndpoint() {
+    // The default endpoint is resolved by the client library, which honors
+    // GOOGLE_CLOUD_UNIVERSE_DOMAIN. Only assert the public default if it is not set.
+    assumeTrue(System.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN") == null);
+    Dataflow client = DataflowTransport.newDataflowClient(options).build();
+    assertEquals(Dataflow.DEFAULT_BASE_URL, client.getBaseUrl());
+  }
+
+  @Test
+  public void testDefaultEndpointHonorsUniverseDomain() {
+    Dataflow client =
+        DataflowTransport.newDataflowClient(options).setUniverseDomain(UNIVERSE_DOMAIN).build();
+    assertEquals("https://dataflow." + UNIVERSE_DOMAIN + "/", client.getBaseUrl());
+  }
+
+  @Test
+  public void testRelativeDataflowEndpointHonorsUniverseDomain() {
+    options.setDataflowEndpoint("v0neverExisted");
+    Dataflow client =
+        DataflowTransport.newDataflowClient(options).setUniverseDomain(UNIVERSE_DOMAIN).build();
+    assertEquals("https://dataflow." + UNIVERSE_DOMAIN + "/v0neverExisted/", client.getBaseUrl());
+  }
+
+  @Test
+  public void testAbsoluteDataflowEndpointOverridesUniverseDomain() {
+    options.setDataflowEndpoint("https://custom-dataflow.example.com:8443/some/path/");
+    Dataflow client =
+        DataflowTransport.newDataflowClient(options).setUniverseDomain(UNIVERSE_DOMAIN).build();
+    assertEquals("https://custom-dataflow.example.com:8443/", client.getRootUrl());
+    assertEquals("some/path/", client.getServicePath());
+  }
+
+  @Test
+  public void testApiRootUrlOverridesUniverseDomain() {
+    options.setApiRootUrl("https://custom-dataflow.example.com/");
+    Dataflow client =
+        DataflowTransport.newDataflowClient(options).setUniverseDomain(UNIVERSE_DOMAIN).build();
+    assertEquals("https://custom-dataflow.example.com/", client.getBaseUrl());
+  }
+
+  @Test
+  public void testApiRootUrlWithRelativeDataflowEndpoint() {
+    options.setApiRootUrl("https://custom-dataflow.example.com/");
+    options.setDataflowEndpoint("v0neverExisted");
+    Dataflow client = DataflowTransport.newDataflowClient(options).build();
+    assertEquals("https://custom-dataflow.example.com/v0neverExisted/", client.getBaseUrl());
+  }
+
+  @Test
+  public void testExplicitDefaultApiRootUrlIsNotOverridden() {
+    options.setApiRootUrl(Dataflow.DEFAULT_ROOT_URL);
+    Dataflow client =
+        DataflowTransport.newDataflowClient(options).setUniverseDomain(UNIVERSE_DOMAIN).build();
+    assertEquals(Dataflow.DEFAULT_ROOT_URL, client.getBaseUrl());
+  }
+}