apache/beam · error · IllegalStateException

Inputs to Flatten had incompatible window windowFns: %s, %s

Error message

Inputs to Flatten had incompatible window windowFns: %s, %s

What it means

Flatten merges multiple PCollections into one, so all inputs must share a compatible windowing strategy. Expand compares each input's WindowFn with the first's; incompatible window functions (e.g. fixed 1h windows vs. sliding windows) make the merged output strategy undefined, so it throws IllegalStateException.

Source

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

   * containing all the elements of all the {@link PCollection}s in its input. Implements {@link
   * #pCollections}.
   *
   * @param <T> the type of the elements in the input and output {@code PCollection}s.
   */
  public static class PCollections<T> extends PTransform<PCollectionList<T>, PCollection<T>> {

    private PCollections() {}

    @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();
      }

View on GitHub (pinned to 12126d8942)

Solutions

Re-window one input to match the other using Window.into()/Window.configure() before Flatten
Move Flatten before the point where the branches diverge in windowing, or flatten and then re-window
Ensure both branches derive from a common windowed PCollection or apply identical windowing

Example fix

// before
PCollectionList.of(globalWindowed).and(hourlyWindowed).apply(Flatten.pCollections());
// after
PCollection<String> a2 = globalWindowed.apply(Window.<String>into(FixedWindows.of(Duration.standardHours(1))));
PCollectionList.of(a2).and(hourlyWindowed).apply(Flatten.pCollections());

Defensive patterns

Strategy: validation

Validate before calling

// Validate window fn compatibility before Flatten
WindowingStrategy<?, ?> first = inputs.get(0).getWindowingStrategy();
for (PCollection<?> p : inputs.getAll()) {
  if (!first.getWindowFn().isCompatible(p.getWindowingStrategy().getWindowFn())) {
    throw new IllegalStateException("Incompatible window fns before Flatten: "
        + first.getWindowFn() + " vs " + p.getWindowingStrategy().getWindowFn());
  }
}

Type guard

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

Try / catch

try { merged = inputs.apply(Flatten.pCollections()); }
catch (IllegalStateException e) { // re-window and retry once
  merged = reWindowAll(inputs).apply(Flatten.pCollections());
}

Prevention

Standardize windowing on a shared Window.configure() step all branches pass through
Check windowing strategy compatibility before any Flatten of heterogeneous branches
Re-window immediately after any transform that changes windowing, before merging

When it happens

Trigger: PApply Flatten to PCollections windowed with different WindowFns — e.g. one with FixedWindows.of(1h) and another with SlidingWindows or GlobalWindows.

Common situations: Merging a bounded, globally-windowed collection with an event-time-windowed stream; two branches of a pipeline that re-window differently before Flatten; forgetting to apply withWindowing after a windowing change in one 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

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