Merge pull request apache#15340 from zhoufek/gbk_233

udim · web-flow · commit cf8e08f51bc9 · 2021-08-20T15:48:59.000-07:00
diff --git a/CHANGES.md b/CHANGES.md
@@ -63,7 +63,8 @@
 
 ## Breaking Changes
 
-* X behavior was changed ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
+* Python GBK by defualt will fail on unbounded PCollections that have global windowing and a default trigger. The `--allow_unsafe_triggers` flag can be used to override this. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)).
+* Python GBK will fail if it detects an unsafe trigger unless the `--allow_unsafe_triggers` flag is set. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)).
 
 ## Deprecations
 
@@ -123,7 +124,8 @@
 
 ## Deprecations
 
-* X behavior is deprecated and will be removed in X versions ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)).
+* Python GBK will stop supporting unbounded PCollections that have global windowing and a default trigger in Beam 2.33. This can be overriden with `--allow_unsafe_triggers`. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)).
+* Python GBK will start requiring safe triggers or the `--allow_unsafe_triggers` flag starting with Beam 2.33. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)).
 
 ## Known Issues
 
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
@@ -534,8 +534,7 @@ def _add_argparse_args(cls, parser):
         'compatibility. See BEAM-11719.')
     parser.add_argument(
         '--allow_unsafe_triggers',
-        # TODO(BEAM-9487): Set to False for Beam 2.33
-        default=True,
+        default=False,
         action='store_true',
         help='Allow the use of unsafe triggers. Unsafe triggers have the '
         'potential to cause data loss due to finishing and/or never having '
diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py
@@ -2320,10 +2320,10 @@ def expand(self, pcoll):
       if pcoll.pipeline.allow_unsafe_triggers:
         # TODO(BEAM-9487) Change comment for Beam 2.33
         _LOGGER.warning(
-            'PCollection passed to GroupByKey (label: %s) is unbounded, has a '
-            'global window, and uses a default trigger. This will no longer '
-            'be allowed starting with Beam 2.33 unless '
-            '--allow_unsafe_triggers is set.',
+            '%s: PCollection passed to GroupByKey is unbounded, has a global '
+            'window, and uses a default trigger. This is being allowed '
+            'because --allow_unsafe_triggers is set, but it may prevent '
+            'data from making it through the pipeline.',
             self.label)
       else:
         raise ValueError(
@@ -2332,22 +2332,19 @@ def expand(self, pcoll):
 
     unsafe_reason = trigger.may_lose_data(windowing)
     if unsafe_reason != DataLossReason.NO_POTENTIAL_LOSS:
+      reason_msg = str(unsafe_reason).replace('DataLossReason.', '')
       if pcoll.pipeline.allow_unsafe_triggers:
-        # TODO(BEAM-9487): Switch back to this log for Beam 2.33.
-        # _LOGGER.warning(
-        #   'Skipping trigger safety check. '
-        #   'This could lead to incomplete or missing groups.')
         _LOGGER.warning(
-            '%s: Unsafe trigger type (%s) detected. Starting with '
-            'Beam 2.33, this will raise an error by default. '
-            'Either change the pipeline to use a safe trigger or '
-            'set the --allow_unsafe_triggers flag.',
+            '%s: Unsafe trigger `%s` detected (reason: %s). This is '
+            'being allowed because --allow_unsafe_triggers is set. This could '
+            'lead to missing or incomplete groups.',
             self.label,
-            unsafe_reason)
+            trigger,
+            reason_msg)
       else:
-        msg = 'Unsafe trigger: `{}` may lose data. '.format(trigger)
-        msg += 'Reason: {}. '.format(
-            str(unsafe_reason).replace('DataLossReason.', ''))
+        msg = '{}: Unsafe trigger: `{}` may lose data. '.format(
+            self.label, trigger)
+        msg += 'Reason: {}. '.format(reason_msg)
         msg += 'This can be overriden with the --allow_unsafe_triggers flag.'
         raise ValueError(msg)
 
