Log temp/staging location options parsing errors. (#39305)

* Log temp/staging location options parsing errors.

* yapf

Author: tvalentyn

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:

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 []