{"record":{"id":"0f37e2123114ebbb","repo":"apache/beam","slug":"inputs-to-flatten-had-incompatible-triggers-s-s","errorCode":null,"errorMessage":"Inputs to Flatten had incompatible triggers: %s, %s","messagePattern":"Inputs to Flatten had incompatible triggers: (.+?), (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Flatten.java","lineNumber":189,"sourceCode":"\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      }\n\n      return PCollection.createPrimitiveOutputInternal(\n          inputs.getPipeline(),\n          windowingStrategy,\n          isBounded,\n          // Take coder from first collection. If there are none, will be left unspecified.\n          inputs.getAll().isEmpty() ? null : inputs.get(0).getCoder());\n    }","sourceCodeStart":171,"sourceCodeEnd":207,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Flatten.java#L171-L207","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Apply identical triggering configuration to all inputs before Flatten","Reset one branch to the default trigger (or the shared trigger) via Window.triggering(...) with the same settings","Flatten before applying divergent trigger configuration, then trigger the merged result"],"exampleFix":"// before\nPCollectionList.of(defaultTriggered).and(triggered).apply(Flatten.pCollections());\n// after\nPCollection<String> a2 = defaultTriggered.apply(Window.<String>configure()\n    .triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow())).discardingFiredPanes());\nPCollectionList.of(a2).and(triggered).apply(Flatten.pCollections());","handlingStrategy":"validation","validationCode":"// Validate trigger compatibility before Flatten\nWindowingStrategy<?, ?> first = inputs.get(0).getWindowingStrategy();\nfor (PCollection<?> p : inputs.getAll()) {\n  if (!first.getTrigger().isCompatible(p.getWindowingStrategy().getTrigger())) {\n    throw new IllegalStateException(\"Incompatible triggers before Flatten\");\n  }\n}","typeGuard":"boolean triggersCompatible(PCollectionList<?> list) {\n  WindowingStrategy<?, ?> s = list.get(0).getWindowingStrategy();\n  return list.getAll().stream().allMatch(p ->\n      s.getTrigger().isCompatible(p.getWindowingStrategy().getTrigger()));\n}","tryCatchPattern":"try { merged = inputs.apply(Flatten.pCollections()); }\ncatch (IllegalStateException e) { // align triggers then flatten\n  merged = alignTriggers(inputs).apply(Flatten.pCollections());\n}","preventionTips":["If one branch uses custom triggering, apply matching triggering to every branch feeding the Flatten","Keep trigger configuration in a shared helper used by all pipeline branches","Flatten before custom trigger configuration, then trigger the merged output"],"tags":["java","apache-beam","windowing","triggers","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"}