This is an automated email from the ASF dual-hosted git repository.
Abacn 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 c2c910a70bf Fix Dataflow ValueProvider serialization (#39614)
c2c910a70bf is described below
commit c2c910a70bfe37926b202b39d7e8b1f21a8a7ccf
Author: Bruno Volpato <[email protected]>
AuthorDate: Wed Aug 5 10:02:01 2026 -0400
Fix Dataflow ValueProvider serialization (#39614)
---
CHANGES.md | 1 +
.../runners/dataflow/internal/apiclient.py | 6 ++++--
.../runners/dataflow/internal/apiclient_test.py | 19 +++++++++++++++++++
3 files changed, 24 insertions(+), 2 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index b98c46cb881..0c4f71aa5a5 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -115,6 +115,7 @@
## Bugfixes
+* Fixed unresolved runtime `ValueProvider` options being stringified in Python
Dataflow Flex Templates ([#39499](https://github.com/apache/beam/issues/39499)).
* Fixed unbounded checkpoint state growth for splittable DoFns that
self-checkpoint on the portable Flink runner (Java)
([#27648](https://github.com/apache/beam/issues/27648)).
* Improved Java pipeline performance by avoiding repeated `DoFn` type
descriptor resolution when creating cached invokers
([#39309](https://github.com/apache/beam/issues/39309)).
* (Python) Fixed a memory leak in Python SDK caused by storing exceptions with
potentially large stack frames in a cache
([#39406](https://github.com/apache/beam/issues/39406)).
diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
index ac411864310..0875bdd14df 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
@@ -280,8 +280,10 @@ class Environment(object):
for k, v in sdk_pipeline_options.items():
if v is None:
continue
- options_dict[k] = str(v) if isinstance(
- v, value_provider.ValueProvider) else v
+ if isinstance(v, value_provider.ValueProvider):
+ options_dict[k] = v.get() if v.is_accessible() else None
+ else:
+ options_dict[k] = v
options_dict["pipelineUrl"] = proto_pipeline_staged_url
if pipeline_proto_hash:
options_dict["pipelineProtoHash"] = pipeline_proto_hash
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 dc55a28cecf..4fca13abee9 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
@@ -113,6 +113,25 @@ class UtilTest(unittest.TestCase):
self.assertEqual(pipeline_url, FAKE_PIPELINE_URL)
+ def test_value_provider_options_serialization(self):
+ class UserOptions(PipelineOptions):
+ @classmethod
+ def _add_argparse_args(cls, parser):
+ parser.add_value_provider_argument('--at_vp_arg1')
+ parser.add_value_provider_argument('--at_vp_arg2')
+
+ pipeline_options = UserOptions([
+ '--at_vp_arg2', 'provided', '--temp_location', 'gs://any-location/temp'
+ ])
+ env = apiclient.Environment([],
+ pipeline_options,
+ '2.0.0',
+ FAKE_PIPELINE_URL)
+
+ recovered_options = env.proto.sdk_pipeline_options['options']
+ self.assertIsNone(recovered_options['at_vp_arg1'])
+ self.assertEqual(recovered_options['at_vp_arg2'], 'provided')
+
def test_pipeline_proto_hash(self):
pipeline_options = PipelineOptions(
['--temp_location', 'gs://any-location/temp'])
