From f70c592f44b65648191f49584cdd1b4b3f10443f Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Mon, 13 Jul 2026 09:22:21 -0700 Subject: [PATCH 1/2] Log temp/staging location options parsing errors. --- .../apache_beam/options/pipeline_options.py | 15 ++++++++++++--- .../options/pipeline_options_validator.py | 7 ++++++- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py index 4b06b8eda613..b3dbeaa741f6 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -1243,18 +1243,27 @@ def _warn_if_soft_delete_policy_enabled(self, arg_name): key="message") except ImportError: _LOGGER.warning('Unable to check soft delete policy due to import error.') - # If either temp or staging location has an issue, we use the valid one for # both locations. If both are bad we return an error. def _handle_temp_and_staging_locations(self, validator): temp_errors = validator.validate_gcs_path(self, 'temp_location') staging_errors = validator.validate_gcs_path(self, 'staging_location') + + temp_location = getattr(self, 'temp_location', None) + staging_location = getattr(self, 'staging_location', None) + + if temp_location is not None and temp_errors: + _LOGGER.warning(temp_errors[0]) + + if staging_location is not None and staging_errors: + _LOGGER.warning(staging_errors[0]) + if temp_errors and not staging_errors: - setattr(self, 'temp_location', getattr(self, 'staging_location')) + setattr(self, 'temp_location', staging_location) self._warn_if_soft_delete_policy_enabled('staging_location') return [] elif staging_errors and not temp_errors: - setattr(self, 'staging_location', getattr(self, 'temp_location')) + setattr(self, 'staging_location', temp_location) self._warn_if_soft_delete_policy_enabled('temp_location') return [] elif not staging_errors and not temp_errors: diff --git a/sdks/python/apache_beam/options/pipeline_options_validator.py b/sdks/python/apache_beam/options/pipeline_options_validator.py index 29253329908e..28786f3c8933 100644 --- a/sdks/python/apache_beam/options/pipeline_options_validator.py +++ b/sdks/python/apache_beam/options/pipeline_options_validator.py @@ -203,7 +203,12 @@ def validate_gcs_path(self, view, arg_name): if not self.is_full_string_match(self.GCS_BUCKET, bucket): return self._validate_error(self.ERR_INVALID_GCS_BUCKET, arg, arg_name) - if gcs_object is None or '\n' in gcs_object or '\r' in gcs_object: + if gcs_object is None: + return self._validate_error( + "Invalid GCS path: '%s' given for option: %s. " + "Did you mean: 'gs://%s/ or gs://some_bucket/%s'?", + arg, arg_name, bucket, bucket) + if '\n' in gcs_object or '\r' in gcs_object: return self._validate_error(self.ERR_INVALID_GCS_OBJECT, arg, arg_name) return [] From fa50f4c721892bcd35ccb6092b30b9eccc06fb5c Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Mon, 13 Jul 2026 10:29:42 -0700 Subject: [PATCH 2/2] yapf --- sdks/python/apache_beam/options/pipeline_options.py | 1 + .../python/apache_beam/options/pipeline_options_validator.py | 5 ++++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py index b3dbeaa741f6..afbec3f46f71 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -1243,6 +1243,7 @@ def _warn_if_soft_delete_policy_enabled(self, arg_name): key="message") except ImportError: _LOGGER.warning('Unable to check soft delete policy due to import error.') + # If either temp or staging location has an issue, we use the valid one for # both locations. If both are bad we return an error. def _handle_temp_and_staging_locations(self, validator): diff --git a/sdks/python/apache_beam/options/pipeline_options_validator.py b/sdks/python/apache_beam/options/pipeline_options_validator.py index 28786f3c8933..9b0d2e6109d6 100644 --- a/sdks/python/apache_beam/options/pipeline_options_validator.py +++ b/sdks/python/apache_beam/options/pipeline_options_validator.py @@ -207,7 +207,10 @@ def validate_gcs_path(self, view, arg_name): return self._validate_error( "Invalid GCS path: '%s' given for option: %s. " "Did you mean: 'gs://%s/ or gs://some_bucket/%s'?", - arg, arg_name, bucket, bucket) + arg, + arg_name, + bucket, + bucket) if '\n' in gcs_object or '\r' in gcs_object: return self._validate_error(self.ERR_INVALID_GCS_OBJECT, arg, arg_name) return []