{"record":{"id":"65ce70ec38a812dd","repo":"apache/beam","slug":"unsafe-trigger-may-lose-data-reason-this-can-be-overriden","errorCode":null,"errorMessage":"{}: Unsafe trigger: `{}` may lose data. Reason: {}. This can be overriden with the --allow_unsafe_triggers flag.","messagePattern":"(.+?): Unsafe trigger: `(.+?)` may lose data\\. Reason: (.+?)\\. This can be overriden with the --allow_unsafe_triggers flag\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/core.py","lineNumber":3537,"sourceCode":"            'PCollection with global windowing and a default trigger')\n\n    unsafe_reason = trigger.may_lose_data(windowing)\n    if unsafe_reason != DataLossReason.NO_POTENTIAL_LOSS:\n      reason_msg = str(unsafe_reason).replace('DataLossReason.', '')\n      if pcoll.pipeline.allow_unsafe_triggers:\n        _LOGGER.warning(\n            '%s: Unsafe trigger `%s` detected (reason: %s). This is '\n            'being allowed because --allow_unsafe_triggers is set. This could '\n            'lead to missing or incomplete groups.',\n            self.label,\n            trigger,\n            reason_msg)\n      else:\n        msg = '{}: Unsafe trigger: `{}` may lose data. '.format(\n            self.label, trigger)\n        msg += 'Reason: {}. '.format(reason_msg)\n        msg += 'This can be overriden with the --allow_unsafe_triggers flag.'\n        raise ValueError(msg)\n\n    return pvalue.PCollection.from_(pcoll)\n\n  def infer_output_type(self, input_type):\n    key_type, value_type = (typehints.typehints.coerce_to_kv_type(\n        input_type).tuple_types)\n    return typehints.KV[key_type, typehints.Iterable[value_type]]\n\n  def to_runner_api_parameter(self, unused_context):\n    # type: (PipelineContext) -> tuple[str, typing.Optional[typing.Union[message.Message, bytes, str]]]\n    # if we're containing a GroupByEncryptedKey, don't allow runners to\n    # recognize this transform as a GBEK so that it doesn't get replaced.\n    if self._replaced_by_gbek:\n      return super().to_runner_api_parameter(unused_context)\n    return common_urns.primitives.GROUP_BY_KEY.urn, None\n\n  @staticmethod\n  @PTransform.register_urn(common_urns.primitives.GROUP_BY_KEY.urn, None)","sourceCodeStart":3519,"sourceCodeEnd":3555,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/core.py#L3519-L3555","documentation":"Beam's GroupByKey validates that the configured trigger does not potentially discard data (may_lose_data returns NO_POTENTIAL_LOSS). If the trigger may lose data, this ValueError is raised naming the transform label, the trigger, and the specific DataLossReason, and pointing to the --allow_unsafe_triggers escape hatch.","triggerScenarios":"Applying GroupByKey on a windowed streaming PCollection whose WindowInto uses a trigger such as a one-shot AfterCount or AfterWatermark without allowed lateness/accumulation that Beam deems unsafe — any trigger where may_lose_data() != NO_POTENTIAL_LOSS.","commonSituations":"Custom triggers built with beam.trigger.* (e.g. AfterProcessingTime with unbounded lateness, element-count triggers) combined with grouping; a runner upgrade starts surfacing the check on previously-accepted pipelines.","solutions":["Redesign the trigger to be safe, e.g. use Repeatedly/AfterWatermark with allowed lateness and with_accumulation_mode(ACCUMULATING)","Increase allowed_lateness in WindowInto so late data is not discarded","Pass --allow_unsafe_triggers to override, only after accepting possible data loss","Log the reason_msg (from DataLossReason) to understand exactly which data can be lost"],"exampleFix":"// before\nbeam.WindowInto(FixedWindows(60), trigger=AfterCount(1))\n// after\nbeam.WindowInto(FixedWindows(60),\n  trigger=Repeatedly(AfterWatermark(late=AfterCount(1))),\n  accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING,\n  allowed_lateness=300)","handlingStrategy":"validation","validationCode":"from apache_beam.transforms.trigger import DataLossReason\nunsafe = trigger.may_lose_data(windowing)\nassert unsafe == DataLossReason.NO_POTENTIAL_LOSS, f'trigger may lose data: {unsafe}'","typeGuard":"def is_safe_trigger(trigger, windowing) -> bool:\n    from apache_beam.transforms.trigger import DataLossReason\n    return trigger.may_lose_data(windowing) == DataLossReason.NO_POTENTIAL_LOSS","tryCatchPattern":"try:\n    expanded = pcoll | beam.GroupByKey()\nexcept ValueError as e:\n    if 'Unsafe trigger' in str(e):\n        expanded = (pcoll | beam.WindowInto(\n            beam.window.FixedWindows(60),\n            allowed_lateness=300,\n            accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING)\n            | beam.GroupByKey())\n    else:\n        raise","preventionTips":["Check trigger.may_lose_data(windowing) when designing custom triggers","Prefer Repeatedly/AfterWatermark with allowed lateness over one-shot triggers","Treat --allow_unsafe_triggers as a last-resort diagnostic flag, not a production setting"],"tags":["python","apache-beam","trigger","data-loss","streaming"],"backgroundTag":"invalid-config-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}