{"record":{"id":"cbc4cc34b1c0612c","repo":"apache/beam","slug":"distinct-does-not-support-merging-windowing-strategies","errorCode":null,"errorMessage":"Distinct does not support merging windowing strategies, except when using the default trigger and zero allowed lateness.","messagePattern":"Distinct does not support merging windowing strategies, except when using the default trigger and zero allowed lateness\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Distinct.java","lineNumber":94,"sourceCode":"  }\n\n  /**\n   * Returns a {@code Distinct<T, IdT>} {@code PTransform}.\n   *\n   * @param <T> the type of the elements of the input and output {@code PCollection}s\n   * @param <IdT> the type of the representative value used to dedup\n   */\n  public static <T, IdT> WithRepresentativeValues<T, IdT> withRepresentativeValueFn(\n      SerializableFunction<T, IdT> fn) {\n    return new WithRepresentativeValues<>(fn, null);\n  }\n\n  private static <T, W extends BoundedWindow> void validateWindowStrategy(\n      WindowingStrategy<T, W> strategy) {\n    if (strategy.needsMerge()\n        && (!strategy.getTrigger().getClass().equals(DefaultTrigger.class)\n            || strategy.getAllowedLateness().isLongerThan(Duration.ZERO))) {\n      throw new UnsupportedOperationException(\n          String.format(\n              \"%s does not support merging windowing strategies, except when using the default \"\n                  + \"trigger and zero allowed lateness.\",\n              Distinct.class.getSimpleName()));\n    }\n  }\n\n  @Override\n  public PCollection<T> expand(PCollection<T> in) {\n    validateWindowStrategy(in.getWindowingStrategy());\n    PCollection<KV<T, Void>> combined =\n        in.apply(\n                \"KeyByElement\",\n                MapElements.via(\n                    new SimpleFunction<T, KV<T, Void>>() {\n                      @Override\n                      public KV<T, Void> apply(T element) {\n                        return KV.of(element, (Void) null);","sourceCodeStart":76,"sourceCodeEnd":112,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Distinct.java#L76-L112","documentation":"Distinct.withRepresentativeValueFn / Distinct of keyed representative values cannot run on a PCollection whose WindowingStrategy requires window merging (e.g. sliding/custom windows) unless the trigger is the DefaultTrigger and allowed lateness is zero. Deduplication with representative values relies on grouping that assumes windows don't merge, so validateWindowStrategy rejects the strategy at graph expansion time.","triggerScenarios":"Applying Distinct (or Distinct.withRepresentativeValueFn) to a PCollection windowed with merging windows (e.g. SlidingWindows or Sessions) while the trigger is not DefaultTrigger or allowedLateness > Duration.ZERO.","commonSituations":"Deduplicating event data in sliding or session windows with custom triggers/allowed lateness set for late data handling.","solutions":["Use the default trigger and set allowed lateness to zero for the windowing before Distinct","Switch to non-merging windows (e.g. FixedWindows) if the semantics allow","Deduplicate before windowing (apply Distinct on the unwindowed/unbounded stream stage), or dedupe via a keyed Combine/GroupByKey approach that supports merging windows"],"exampleFix":"// before\npc.apply(Window.<T>into(SlidingWindows.of(Duration.standardMinutes(5)).every(Duration.standardMinutes(1))).triggering(AfterWatermark.pastEndOfWindow()).withAllowedLateness(StandardMinutes(1)))\n   .apply(Distinct.<T>create());\n// after\npc.apply(Distinct.<T>create())\n  .apply(Window.<T>into(SlidingWindows.of(Duration.standardMinutes(5)).every(Duration.standardMinutes(1))));","handlingStrategy":"validation","validationCode":"WindowingStrategy<?, ?> ws = pc.getWindowingStrategy();\nif (ws.needsMerge() && !(ws.getTrigger() instanceof DefaultTrigger)\n    && ws.getAllowedLateness().isLongerThan(Duration.ZERO)) {\n  // restructure windowing before applying Distinct\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Remember Distinct with merging windows requires DefaultTrigger and zero lateness","Dedupe before applying sliding/session windowing when possible"],"tags":["apache-beam","java","windowing","unsupported-operation"],"backgroundTag":"unsupported-operation","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"}