apache/beam · error · IllegalArgumentException

Unsafe trigger ' ' may lose data, did you mean to wrap it…

Error message

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.

What it means

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.

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

Example fix

// before
Window.into(FixedWindows.of(Duration.standardMinutes(1)))
    .triggering(AfterWatermark.pastEndOfWindow());
// after
Window.into(FixedWindows.of(Duration.standardMinutes(1)))
    .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardSeconds(30))));

Defensive patterns

Strategy: validation

Validate before calling

if (!(windowing.getTrigger() instanceof Repeatedly)
    && !(windowing.getTrigger() instanceof DefaultTrigger)) {
  // wrap finishing triggers before grouping
  triggering = Repeatedly.forever(triggering);
}

Try / catch

try {
  grouped = input.apply(GroupByKey.create());
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("Unsafe trigger")) { /* wrap trigger in Repeatedly.forever */ }
  else throw e;
}

Prevention

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

When it happens

Trigger: 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(...).

Common situations: 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.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/bd3e8ae680620794. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java:187

  }

  /////////////////////////////////////////////////////////////////////////////

  public static void applicableTo(PCollection<?> input) {
    WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
    // Verify that the input PCollection is bounded, or that there is windowing/triggering being
    // used. Without this, the watermark (at end of global window) will never be reached.
    if (windowingStrategy.getWindowFn() instanceof GlobalWindows
        && windowingStrategy.getTrigger() instanceof DefaultTrigger
        && input.isBounded() != IsBounded.BOUNDED) {
      throw new IllegalStateException(
          "GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without a"
              + " trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.");
    }

    // Validate that the trigger does not finish before garbage collection time
    if (!triggerIsSafe(windowingStrategy)) {
      throw new IllegalArgumentException(
          String.format(
              "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.",
              windowingStrategy.getTrigger()));
    }
  }

  @Override
  public void validate(
      @Nullable PipelineOptions options,
      Map<TupleTag<?>, PCollection<?>> inputs,
      Map<TupleTag<?>, PCollection<?>> outputs) {
    PCollection<?> input = Iterables.getOnlyElement(inputs.values());
    KvCoder<K, V> inputCoder = getInputKvCoder(input.getCoder());

    // Ensure that the output coder key and value types aren't different.

View on GitHub (pinned to 12126d8942)