diff --git a/CHANGES.md b/CHANGES.md index 9cbbe1c207fb..fcfb10c08bbd 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -112,6 +112,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 ac4118643109..0875bdd14df5 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 @@ def __init__( 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 dc55a28cecf4..4fca13abee99 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 @@ def test_pipeline_url(self): 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'])