This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new d4ba44da40b [Java] Honor GOOGLE_CLOUD_UNIVERSE_DOMAIN for the Dataflow
endpoint (#40672)
d4ba44da40b is described below
commit d4ba44da40b4a638fda3d92ae87caab26b92da31
Author: Shunping Huang <[email protected]>
AuthorDate: Thu Oct 22 18:43:47 2026 -0400
[Java] Honor GOOGLE_CLOUD_UNIVERSE_DOMAIN for the Dataflow endpoint (#40672)
---
CHANGES.md | 1 +
.../options/DataflowPipelineDebugOptions.java | 17 +++-
.../dataflow/options/DataflowPipelineOptions.java | 3 +-
.../runners/dataflow/util/DataflowTransport.java | 52 ++++++----
.../dataflow/util/DataflowTransportTest.java | 109 +++++++++++++++++++++
5 files changed, 159 insertions(+), 23 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 77f03f3da76..39d25b568a0 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -93,6 +93,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)).
diff --git
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java
index 7a5284151b9..caa27c546ea 100644
---
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java
+++
b/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.Hidden;
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,13 +48,20 @@ 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);
@@ -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"
diff --git
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java
index 57f927d7307..02ae3bc762b 100644
---
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java
+++
b/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"
diff --git
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTransport.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTransport.java
index ad579333ebf..a93c3dec25d 100644
---
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTransport.java
+++
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTransport.java
@@ -30,7 +30,9 @@ import java.net.URL;
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 @@ public class DataflowTransport {
}
}
- /** 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(
diff --git
a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/DataflowTransportTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/util/DataflowTransportTest.java
new file mode 100644
index 00000000000..1cb34430a21
--- /dev/null
+++
b/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());
+ }
+}
