apache/beam · error · IllegalStateException

Inputs to Flatten had incompatible triggers: %s, %s

Error message

Inputs to Flatten had incompatible triggers: %s, %s

What it means

Beyond window functions, Flatten requires all input PCollections to have compatible triggers. If the first input's trigger is not compatible with another input's (e.g. DefaultTrigger vs. Repeatedly.forever(AfterCount)), the merged strategy is ambiguous and expand() throws IllegalStateException.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Flatten.java:189

    @Override
    public PCollection<T> expand(PCollectionList<T> inputs) {
      WindowingStrategy<?, ?> windowingStrategy;
      IsBounded isBounded = IsBounded.BOUNDED;
      if (!inputs.getAll().isEmpty()) {
        windowingStrategy = inputs.get(0).getWindowingStrategy();
        for (PCollection<?> input : inputs.getAll()) {
          WindowingStrategy<?, ?> other = input.getWindowingStrategy();
          if (!windowingStrategy.getWindowFn().isCompatible(other.getWindowFn())) {
            throw new IllegalStateException(
                "Inputs to Flatten had incompatible window windowFns: "
                    + windowingStrategy.getWindowFn()
                    + ", "
                    + other.getWindowFn());
          }

          if (!windowingStrategy.getTrigger().isCompatible(other.getTrigger())) {
            throw new IllegalStateException(
                "Inputs to Flatten had incompatible triggers: "
                    + windowingStrategy.getTrigger()
                    + ", "
                    + other.getTrigger());
          }
          isBounded = isBounded.and(input.isBounded());
        }
      } else {
        windowingStrategy = WindowingStrategy.globalDefault();
      }

      return PCollection.createPrimitiveOutputInternal(
          inputs.getPipeline(),
          windowingStrategy,
          isBounded,
          // Take coder from first collection. If there are none, will be left unspecified.
          inputs.getAll().isEmpty() ? null : inputs.get(0).getCoder());
    }

View on GitHub (pinned to 12126d8942)

Solutions

1. Apply identical triggering configuration to all inputs before Flatten
2. Reset one branch to the default trigger (or the shared trigger) via Window.triggering(...) with the same settings
3. Flatten before applying divergent trigger configuration, then trigger the merged result

Example fix

// before
PCollectionList.of(defaultTriggered).and(triggered).apply(Flatten.pCollections());
// after
PCollection<String> a2 = defaultTriggered.apply(Window.<String>configure()
    .triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow())).discardingFiredPanes());
PCollectionList.of(a2).and(triggered).apply(Flatten.pCollections());

// Validate trigger compatibility before Flatten
WindowingStrategy<?, ?> first = inputs.get(0).getWindowingStrategy();
for (PCollection<?> p : inputs.getAll()) {
  if (!first.getTrigger().isCompatible(p.getWindowingStrategy().getTrigger())) {
    throw new IllegalStateException("Incompatible triggers before Flatten");
  }
}

boolean triggersCompatible(PCollectionList<?> list) {
  WindowingStrategy<?, ?> s = list.get(0).getWindowingStrategy();
  return list.getAll().stream().allMatch(p ->
      s.getTrigger().isCompatible(p.getWindowingStrategy().getTrigger()));
}

try { merged = inputs.apply(Flatten.pCollections()); }
catch (IllegalStateException e) { // align triggers then flatten
  merged = alignTriggers(inputs).apply(Flatten.pCollections());
}

When it happens

Trigger: Flattening PCollections where one has custom triggering (e.g. via Window.triggering(...) or after a GroupByKey with custom trigger) and another uses the default trigger, or two different custom triggers.

Common situations: One pipeline branch sets explicit triggers for latency tuning while the other keeps defaults; reusing a trigger-configured PCollection in a Flatten with a fresh branch.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• Inputs to Flatten had incompatible window windowFns: %s, %s
• Except when using GlobalWindows, calling .triggering() to sp
• Calling .triggering() to specify a trigger or calling .withA
• Distinct does not support merging windowing strategies, exce
• GroupByKey cannot be applied to non-bounded PCollection in t

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