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. * *
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 57f927d73073..02ae3bc762b8 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. * *
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 ad579333ebf4..a93c3dec25d6 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 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. + * + *
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 000000000000..1cb34430a212 --- /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()); + } +}