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 1a017df4529 [Python] Don't override the Dataflow endpoint by default
(#40568)
1a017df4529 is described below
commit 1a017df4529bc6b005eaa2a4d12ddb63fa2569f5
Author: Shunping Huang <[email protected]>
AuthorDate: Thu Oct 15 09:05:15 2026 -0400
[Python] Don't override the Dataflow endpoint by default (#40568)
* [Python] Don't override the Dataflow endpoint by default
The default --dataflow_endpoint (GDU URL) was always passed as an
api_endpoint override, ignoring GOOGLE_CLOUD_UNIVERSE_DOMAIN (e.g. TPC).
Default it to '' so the client library resolves the endpoint, and drop
the legacy TaskRunner worker settings that hard-coded the GDU URL.
* Reformat
* Edited CHANGES.md
* Print out the actual endpoint where the job request is sent to.
* Reword comments
---
CHANGES.md | 1 +
.../python/apache_beam/options/pipeline_options.py | 8 ++-
.../options/pipeline_options_validator.py | 13 ++--
.../runners/dataflow/internal/apiclient.py | 9 +--
.../runners/dataflow/internal/apiclient_test.py | 70 +++++++++++++++++-----
5 files changed, 70 insertions(+), 31 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index f98790bc2e1..14c7e676f18 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -84,6 +84,7 @@
`jakarta.jms.*`. The module now requires Java 17 at runtime
([#38729](https://github.com/apache/beam/issues/38729)).
* (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)).
## Deprecations
diff --git a/sdks/python/apache_beam/options/pipeline_options.py
b/sdks/python/apache_beam/options/pipeline_options.py
index 84c0743ceaa..08ba34dabee 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -1038,10 +1038,12 @@ class GoogleCloudOptions(PipelineOptions):
def _add_argparse_args(cls, parser):
parser.add_argument(
'--dataflow_endpoint',
- default=cls.DATAFLOW_ENDPOINT,
+ default='',
help=(
- 'The URL for the Dataflow API. If not set, the default public URL '
- 'will be used.'))
+ 'The URL for the Dataflow API. If not set, the endpoint is '
+ 'resolved by the Google Cloud client library, which honors '
+ 'GOOGLE_CLOUD_UNIVERSE_DOMAIN and otherwise uses the default '
+ 'public URL (%s).' % cls.DATAFLOW_ENDPOINT))
# Remote execution must check that this option is not None.
parser.add_argument(
'--project',
diff --git a/sdks/python/apache_beam/options/pipeline_options_validator.py
b/sdks/python/apache_beam/options/pipeline_options_validator.py
index 9b0d2e6109d..54505154641 100644
--- a/sdks/python/apache_beam/options/pipeline_options_validator.py
+++ b/sdks/python/apache_beam/options/pipeline_options_validator.py
@@ -168,9 +168,9 @@ class PipelineOptionsValidator(object):
dataflow_endpoint = (
self.options.view_as(GoogleCloudOptions).dataflow_endpoint)
- if dataflow_endpoint is None:
- return False
- else:
+ # An unset endpoint means the default Dataflow service endpoint,
+ # so only an explicit localhost endpoint opts out.
+ if dataflow_endpoint:
endpoint_parts = urlparse(dataflow_endpoint, allow_fragments=False)
if endpoint_parts.netloc.startswith("localhost"):
return False
@@ -274,10 +274,9 @@ class PipelineOptionsValidator(object):
else:
view.region = default_region
dataflow_endpoint = view.dataflow_endpoint
- if dataflow_endpoint is None:
- errors.extend(
- self._validate_error(self.ERR_MISSING_OPTION, dataflow_endpoint))
- else:
+ # When dataflow_endpoint is unset, the client library resolves
+ # the endpoint (honoring GOOGLE_CLOUD_UNIVERSE_DOMAIN).
+ if dataflow_endpoint:
valid_endpoint = self.validate_endpoint_url(dataflow_endpoint)
if valid_endpoint is False:
errors.extend(
diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
index 654808373f2..8a478f3fb63 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
@@ -180,12 +180,7 @@ class Environment(object):
name=package))
pool = dataflow.WorkerPool(
- kind='local' if self.local else 'harness',
- packages=package_descriptors,
- taskrunner_settings=dataflow.TaskRunnerSettings(
- parallel_worker_settings=dataflow.WorkerSettings(
- base_url=GoogleCloudOptions.DATAFLOW_ENDPOINT,
- service_path=self.google_cloud_options.dataflow_endpoint)))
+ kind='local' if self.local else 'harness',
packages=package_descriptors)
pool.autoscaling_settings = dataflow.AutoscalingSettings()
# Set worker pool options received through command line.
@@ -902,7 +897,7 @@ class DataflowApplicationClient(object):
'HTTP status %d trying to create job'
' at dataflow service endpoint %s',
e.code,
- self.google_cloud_options.dataflow_endpoint)
+ self._jobs_client.api_endpoint)
_LOGGER.fatal('details of server error: %s', e)
raise
diff --git
a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
index fdce35b4b5e..b82c7f55067 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
@@ -1320,12 +1320,19 @@ class UtilTest(unittest.TestCase):
self.assertFalse(template_obj.get('steps'))
self.assertTrue(template_obj['stepsLocation'])
+ @mock.patch.dict(
+ os.environ, {'GOOGLE_CLOUD_UNIVERSE_DOMAIN': 'custom-universe.com'})
def test_dataflow_endpoint_clean(self):
+ # An explicit endpoint is cleaned and passed as an api_endpoint override to
+ # every Dataflow client, regardless of GOOGLE_CLOUD_UNIVERSE_DOMAIN.
endpoints_and_expectations = [
# (input_endpoint, expected_endpoint, expected_transport)
('https://dataflow.googleapis.com/', 'dataflow.googleapis.com', None),
('https://dataflow.googleapis.com ', 'dataflow.googleapis.com',
None),
('dataflow.googleapis.com/', 'dataflow.googleapis.com', None),
+ ('https://dataflow.example.com/', 'dataflow.example.com', None),
+ ('https://dataflow.example.com ', 'dataflow.example.com', None),
+ ('dataflow.example.com/', 'dataflow.example.com', None),
('http://localhost:8080/', 'http://localhost:8080', 'rest'),
('localhost:8080/', 'localhost:8080', 'rest'),
]
@@ -1341,20 +1348,55 @@ class UtilTest(unittest.TestCase):
'--no_auth',
])
with
mock.patch('apache_beam.runners.dataflow.internal.apiclient.dataflow'
- '.JobsV1Beta3Client') as mock_jobs:
- with mock.patch(
- 'apache_beam.runners.dataflow.internal.apiclient.dataflow'
- '.MessagesV1Beta3Client'):
- with mock.patch(
- 'apache_beam.runners.dataflow.internal.apiclient.dataflow'
- '.MetricsV1Beta3Client'):
- apiclient.DataflowApplicationClient(pipeline_options)
- mock_jobs.assert_called_once()
- called_kwargs = mock_jobs.call_args.kwargs
- client_opts = called_kwargs.get('client_options')
- self.assertIsNotNone(client_opts)
- self.assertEqual(client_opts.api_endpoint, expected_ep)
- self.assertEqual(called_kwargs.get('transport'),
expected_transport)
+ '.JobsV1Beta3Client') as mock_jobs, \
+ mock.patch('apache_beam.runners.dataflow.internal.apiclient.dataflow'
+ '.MessagesV1Beta3Client') as mock_messages, \
+ mock.patch('apache_beam.runners.dataflow.internal.apiclient.dataflow'
+ '.MetricsV1Beta3Client') as mock_metrics:
+ apiclient.DataflowApplicationClient(pipeline_options)
+ for mock_client in (mock_jobs, mock_messages, mock_metrics):
+ mock_client.assert_called_once()
+ called_kwargs = mock_client.call_args.kwargs
+ client_opts = called_kwargs.get('client_options')
+ self.assertIsNotNone(client_opts)
+ self.assertEqual(client_opts.api_endpoint, expected_ep)
+ self.assertEqual(called_kwargs.get('transport'), expected_transport)
+
+ @mock.patch.dict(
+ os.environ,
+ {
+ # Pin mTLS off so endpoint resolution doesn't depend on the host.
+ 'GOOGLE_API_USE_CLIENT_CERTIFICATE': 'false',
+ 'GOOGLE_API_USE_MTLS_ENDPOINT': 'never',
+ })
+ def test_dataflow_endpoint_unset_uses_client_library_default(self):
+ # Without an explicit endpoint, no api_endpoint override is passed, so the
+ # client library resolves the endpoint and honors
+ # GOOGLE_CLOUD_UNIVERSE_DOMAIN.
+ pipeline_options = PipelineOptions([
+ '--project',
+ 'test-project',
+ '--temp_location',
+ 'gs://test-location/temp',
+ '--no_auth',
+ ])
+ self.assertEqual(
+ pipeline_options.view_as(GoogleCloudOptions).dataflow_endpoint, '')
+
+ def resolved_endpoints():
+ client = apiclient.DataflowApplicationClient(pipeline_options)
+ return {
+ c.api_endpoint
+ for c in (
+ client._jobs_client, client._messages_client,
+ client._metrics_client)
+ }
+
+ os.environ.pop('GOOGLE_CLOUD_UNIVERSE_DOMAIN', None)
+ self.assertEqual(resolved_endpoints(), {'dataflow.googleapis.com'})
+
+ os.environ['GOOGLE_CLOUD_UNIVERSE_DOMAIN'] = 'custom-universe.com'
+ self.assertEqual(resolved_endpoints(), {'dataflow.custom-universe.com'})
def test_stage_resources(self):
pipeline_options = PipelineOptions([
