apache/beam · error · UnsupportedOperationException

Distinct does not support merging windowing strategies, exce

Error message

Distinct does not support merging windowing strategies, except when using the default trigger and zero allowed lateness.

What it means

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.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Distinct.java:94

  }

  /**
   * Returns a {@code Distinct<T, IdT>} {@code PTransform}.
   *
   * @param <T> the type of the elements of the input and output {@code PCollection}s
   * @param <IdT> the type of the representative value used to dedup
   */
  public static <T, IdT> WithRepresentativeValues<T, IdT> withRepresentativeValueFn(
      SerializableFunction<T, IdT> fn) {
    return new WithRepresentativeValues<>(fn, null);
  }

  private static <T, W extends BoundedWindow> void validateWindowStrategy(
      WindowingStrategy<T, W> strategy) {
    if (strategy.needsMerge()
        && (!strategy.getTrigger().getClass().equals(DefaultTrigger.class)
            || strategy.getAllowedLateness().isLongerThan(Duration.ZERO))) {
      throw new UnsupportedOperationException(
          String.format(
              "%s does not support merging windowing strategies, except when using the default "
                  + "trigger and zero allowed lateness.",
              Distinct.class.getSimpleName()));
    }
  }

  @Override
  public PCollection<T> expand(PCollection<T> in) {
    validateWindowStrategy(in.getWindowingStrategy());
    PCollection<KV<T, Void>> combined =
        in.apply(
                "KeyByElement",
                MapElements.via(
                    new SimpleFunction<T, KV<T, Void>>() {
                      @Override
                      public KV<T, Void> apply(T element) {
                        return KV.of(element, (Void) null);

View on GitHub (pinned to 12126d8942)

Solutions

  1. Use the default trigger and set allowed lateness to zero for the windowing before Distinct
  2. Switch to non-merging windows (e.g. FixedWindows) if the semantics allow
  3. Deduplicate before windowing (apply Distinct on the unwindowed/unbounded stream stage), or dedupe via a keyed Combine/GroupByKey approach that supports merging windows

Example fix

// before
pc.apply(Window.<T>into(SlidingWindows.of(Duration.standardMinutes(5)).every(Duration.standardMinutes(1))).triggering(AfterWatermark.pastEndOfWindow()).withAllowedLateness(StandardMinutes(1)))
   .apply(Distinct.<T>create());
// after
pc.apply(Distinct.<T>create())
  .apply(Window.<T>into(SlidingWindows.of(Duration.standardMinutes(5)).every(Duration.standardMinutes(1))));
Defensive patterns

Strategy: validation

Validate before calling

WindowingStrategy<?, ?> ws = pc.getWindowingStrategy();
if (ws.needsMerge() && !(ws.getTrigger() instanceof DefaultTrigger)
    && ws.getAllowedLateness().isLongerThan(Duration.ZERO)) {
  // restructure windowing before applying Distinct
}

Prevention

When it happens

Trigger: 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.

Common situations: Deduplicating event data in sliding or session windows with custom triggers/allowed lateness set for late data handling.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

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