diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py index 4b06b8eda613..afbec3f46f71 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -1249,12 +1249,22 @@ def _warn_if_soft_delete_policy_enabled(self, arg_name): 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..9b0d2e6109d6 100644 --- a/sdks/python/apache_beam/options/pipeline_options_validator.py +++ b/sdks/python/apache_beam/options/pipeline_options_validator.py @@ -203,7 +203,15 @@ 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 []