Fix the soft-delete check and emit soft-delete log warning at most once per bucket. (#36585)

* Fix soft-delete check and emit soft-delete log warning at most once per bucket.

* Append a trailing slash to the new bucket name, otherwise soft-delete policy check doesn't work.

* Include bucket name in warning.

* Drive-by: also log unparsed options at most once.

Author: tvalentyn

sdks/python/apache_beam/io/gcp/gcsio.py

@@ -642,7 +642,7 @@ def _updated_to_seconds(updated):
 
   def is_soft_delete_enabled(self, gcs_path):
     try:
-      bucket_name, _ = parse_gcs_path(gcs_path)
+      bucket_name, _ = parse_gcs_path(gcs_path, object_optional=True)
       bucket = self.get_bucket(bucket_name)
       if (bucket.soft_delete_policy is not None and
           bucket.soft_delete_policy.retention_duration_seconds > 0):

sdks/python/apache_beam/options/pipeline_options.py

@@ -38,6 +38,7 @@
 from apache_beam.options.value_provider import StaticValueProvider
 from apache_beam.options.value_provider import ValueProvider
 from apache_beam.transforms.display import HasDisplayData
+from apache_beam.utils import logger
 from apache_beam.utils import proto_utils
 
 __all__ = [
@@ -475,7 +476,7 @@ def _warn_on_unknown_options(unknown_args, parser):
         suggestions = difflib.get_close_matches(arg_name, all_known_options)
         if suggestions:
           msg += f". Did you mean '{suggestions[0]}'?'"
-      _LOGGER.warning(msg)
+      logger.log_first_n(logging.WARN, msg, key="message")
 
   def get_all_options(
       self,
@@ -1171,7 +1172,7 @@ def _create_default_gcs_bucket(self):
       return None
     bucket = gcsio.get_or_create_default_gcs_bucket(self)
     if bucket:
-      return 'gs://%s' % bucket.id
+      return 'gs://%s/' % bucket.id
     else:
       return None
 
@@ -1187,14 +1188,19 @@ def _warn_if_soft_delete_policy_enabled(self, arg_name):
     try:
       from apache_beam.io.gcp import gcsio
       if gcsio.GcsIO().is_soft_delete_enabled(gcs_path):
-        _LOGGER.warning(
-            "Bucket specified in %s has soft-delete policy enabled."
+        logger.log_first_n(
+            logging.WARN,
+            "Bucket %s used as %s has soft-delete policy enabled."
             " To avoid being billed for unnecessary storage costs, turn"
             " off the soft delete feature on buckets that your Dataflow"
             " jobs use for temporary and staging storage. For more"
             " information, see"
             " https://cloud.google.com/storage/docs/use-soft-delete"
-            "#remove-soft-delete-policy." % arg_name)
+            "#remove-soft-delete-policy.",
+            gcs_path,
+            arg_name,
+            n=1,
+            key="message")
     except ImportError:
       _LOGGER.warning('Unable to check soft delete policy due to import error.')