{"record":{"id":"bd3e8ae680620794","repo":"apache/beam","slug":"unsafe-trigger-s-may-lose-data-did-you-mean-to-wrap-it-in","errorCode":null,"errorMessage":"Unsafe trigger '%s' may lose data, did you mean to wrap it in`Repeatedly.forever(...)`?%nSee https://s.apache.org/finishing-triggers-drop-data for details.","messagePattern":"Unsafe trigger '(.+?)' may lose data, did you mean to wrap it in`Repeatedly\\.forever\\(\\.\\.\\.\\)`\\?%nSee https://s\\.apache\\.org/finishing-triggers-drop-data for details\\.","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java","lineNumber":187,"sourceCode":"  }\n\n  /////////////////////////////////////////////////////////////////////////////\n\n  public static void applicableTo(PCollection<?> input) {\n    WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();\n    // Verify that the input PCollection is bounded, or that there is windowing/triggering being\n    // used. Without this, the watermark (at end of global window) will never be reached.\n    if (windowingStrategy.getWindowFn() instanceof GlobalWindows\n        && windowingStrategy.getTrigger() instanceof DefaultTrigger\n        && input.isBounded() != IsBounded.BOUNDED) {\n      throw new IllegalStateException(\n          \"GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without a\"\n              + \" trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.\");\n    }\n\n    // Validate that the trigger does not finish before garbage collection time\n    if (!triggerIsSafe(windowingStrategy)) {\n      throw new IllegalArgumentException(\n          String.format(\n              \"Unsafe trigger '%s' may lose data, did you mean to wrap it in\"\n                  + \"`Repeatedly.forever(...)`?%nSee \"\n                  + \"https://s.apache.org/finishing-triggers-drop-data \"\n                  + \"for details.\",\n              windowingStrategy.getTrigger()));\n    }\n  }\n\n  @Override\n  public void validate(\n      @Nullable PipelineOptions options,\n      Map<TupleTag<?>, PCollection<?>> inputs,\n      Map<TupleTag<?>, PCollection<?>> outputs) {\n    PCollection<?> input = Iterables.getOnlyElement(inputs.values());\n    KvCoder<K, V> inputCoder = getInputKvCoder(input.getCoder());\n\n    // Ensure that the output coder key and value types aren't different.","sourceCodeStart":169,"sourceCodeEnd":205,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java#L169-L205","documentation":"GroupByKey validates that the input's trigger cannot finish (stop accepting data) before window garbage-collection time; a finishing trigger on an unbounded source silently drops data that arrives after the trigger closes. Beam throws IllegalArgumentException in applicableTo() when triggerIsSafe() reports the strategy unsafe.","triggerScenarios":"Applying GroupByKey to a PCollection whose windowing strategy uses a finishing trigger such as AfterAll.of(...), AfterWatermark.pastEndOfWindow().withFinalFiring(...), or a OnceTrigger, without wrapping it in Repeatedly.forever(...).","commonSituations":"Manually composing triggers (e.g. AfterEach.inOrder with a terminal trigger) for streaming aggregation; copying a batch trigger config into a streaming pipeline; a library-provided default that happens to be finishing.","solutions":["Wrap the trigger in Repeatedly.forever(trigger) so it never finishes before GC time","Use AfterWatermark.withEarlyFirings/withLateFirings patterns that don't declare a final finishing firing on unbounded data","Switch to a trigger known to be safe, e.g. Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(...))","If the pipeline is truly batch/bounded, verify the input is bounded — the check still applies to the windowing strategy, so use a non-finishing trigger anyway"],"exampleFix":"// before\nWindow.into(FixedWindows.of(Duration.standardMinutes(1)))\n    .triggering(AfterWatermark.pastEndOfWindow());\n// after\nWindow.into(FixedWindows.of(Duration.standardMinutes(1)))\n    .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardSeconds(30))));","handlingStrategy":"validation","validationCode":"if (!(windowing.getTrigger() instanceof Repeatedly)\n    && !(windowing.getTrigger() instanceof DefaultTrigger)) {\n  // wrap finishing triggers before grouping\n  triggering = Repeatedly.forever(triggering);\n}","typeGuard":null,"tryCatchPattern":"try {\n  grouped = input.apply(GroupByKey.create());\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"Unsafe trigger\")) { /* wrap trigger in Repeatedly.forever */ }\n  else throw e;\n}","preventionTips":["Default to Repeatedly.forever(...) for streaming triggers","Avoid AfterAll/AfterWatermark final firings on unbounded inputs","Review trigger composition when copying batch pipeline configs to streaming"],"tags":["apache-beam","java","triggers","streaming","data-loss"],"backgroundTag":"invalid-argument-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"}