{"record":{"id":"b055879882f55707","repo":"apache/beam","slug":"inputs-to-flatten-had-incompatible-window-windowfns-s-s","errorCode":null,"errorMessage":"Inputs to Flatten had incompatible window windowFns: %s, %s","messagePattern":"Inputs to Flatten had incompatible window windowFns: (.+?), (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Flatten.java","lineNumber":181,"sourceCode":"   * containing all the elements of all the {@link PCollection}s in its input. Implements {@link\n   * #pCollections}.\n   *\n   * @param <T> the type of the elements in the input and output {@code PCollection}s.\n   */\n  public static class PCollections<T> extends PTransform<PCollectionList<T>, PCollection<T>> {\n\n    private PCollections() {}\n\n    @Override\n    public PCollection<T> expand(PCollectionList<T> inputs) {\n      WindowingStrategy<?, ?> windowingStrategy;\n      IsBounded isBounded = IsBounded.BOUNDED;\n      if (!inputs.getAll().isEmpty()) {\n        windowingStrategy = inputs.get(0).getWindowingStrategy();\n        for (PCollection<?> input : inputs.getAll()) {\n          WindowingStrategy<?, ?> other = input.getWindowingStrategy();\n          if (!windowingStrategy.getWindowFn().isCompatible(other.getWindowFn())) {\n            throw new IllegalStateException(\n                \"Inputs to Flatten had incompatible window windowFns: \"\n                    + windowingStrategy.getWindowFn()\n                    + \", \"\n                    + other.getWindowFn());\n          }\n\n          if (!windowingStrategy.getTrigger().isCompatible(other.getTrigger())) {\n            throw new IllegalStateException(\n                \"Inputs to Flatten had incompatible triggers: \"\n                    + windowingStrategy.getTrigger()\n                    + \", \"\n                    + other.getTrigger());\n          }\n          isBounded = isBounded.and(input.isBounded());\n        }\n      } else {\n        windowingStrategy = WindowingStrategy.globalDefault();\n      }","sourceCodeStart":163,"sourceCodeEnd":199,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Flatten.java#L163-L199","documentation":"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.","triggerScenarios":"PApply Flatten to PCollections windowed with different WindowFns — e.g. one with FixedWindows.of(1h) and another with SlidingWindows or GlobalWindows.","commonSituations":"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.","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"],"exampleFix":"// before\nPCollectionList.of(globalWindowed).and(hourlyWindowed).apply(Flatten.pCollections());\n// after\nPCollection<String> a2 = globalWindowed.apply(Window.<String>into(FixedWindows.of(Duration.standardHours(1))));\nPCollectionList.of(a2).and(hourlyWindowed).apply(Flatten.pCollections());","handlingStrategy":"validation","validationCode":"// Validate window fn compatibility before Flatten\nWindowingStrategy<?, ?> first = inputs.get(0).getWindowingStrategy();\nfor (PCollection<?> p : inputs.getAll()) {\n  if (!first.getWindowFn().isCompatible(p.getWindowingStrategy().getWindowFn())) {\n    throw new IllegalStateException(\"Incompatible window fns before Flatten: \"\n        + first.getWindowFn() + \" vs \" + p.getWindowingStrategy().getWindowFn());\n  }\n}","typeGuard":"boolean windowFnsCompatible(PCollectionList<?> list) {\n  WindowingStrategy<?, ?> s = list.get(0).getWindowingStrategy();\n  return list.getAll().stream().allMatch(p ->\n      s.getWindowFn().isCompatible(p.getWindowingStrategy().getWindowFn()));\n}","tryCatchPattern":"try { merged = inputs.apply(Flatten.pCollections()); }\ncatch (IllegalStateException e) { // re-window and retry once\n  merged = reWindowAll(inputs).apply(Flatten.pCollections());\n}","preventionTips":["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"],"tags":["java","apache-beam","windowing","flatten"],"backgroundTag":"incompatible-source-type","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}