apache/beam · error · IllegalArgumentException

PCollections come from different Pipelines

Error message

PCollections come from different Pipelines

What it means

All PCollections in a KeyedPCollectionTuple must belong to the same Pipeline object, since they will be combined into one CoGroupByKey transform. Appending a PCollection from a different Pipeline is rejected.

Solutions

1. Create all inputs and the KeyedPCollectionTuple from the same Pipeline instance
2. If combining data from two pipelines, write one side to a data source and re-read it in the other pipeline before joining
3. Pass the same Pipeline (e.g. p) when constructing both the tuple and the PCollections

Example fix

// before
Pipeline p1 = Pipeline.create(); Pipeline p2 = Pipeline.create();
KeyedPCollectionTuple.of(tag, p1.apply(...)).and(tag2, p2.apply(...)); // throws
// after
KeyedPCollectionTuple.of(tag, p1.apply(...)).and(tag2, p1.apply(...));

if (pc.getPipeline() != tuple.getPipeline()) { throw new IllegalArgumentException("PCollection pipeline mismatch"); }

try { return tuple.and(tag, pc); } catch (IllegalArgumentException e) { throw new IllegalStateException("Cannot join across pipelines", e); }

When it happens

Trigger: Calling keyedTuple.and(tag, pc) where pc was created by a different Pipeline than the tuple's pipeline, e.g. merging results from p1.apply(...) into a tuple built on p2.

Common situations: Two pipelines created in the same test/program; results of an inner pipeline passed to an outer one; reusing builder objects across Pipeline instances.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• cannot return null keyCoder
• DLQ Router only supports PCollectionTuples split between…
• Failed to validate transform
• Illegal access to pipeline after visitor traversal was…
• input must be PBegin or PCollection

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

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/KeyedPCollectionTuple.java:71

  }

  /**
   * A version of {@link #of(TupleTag, PCollection)} that takes in a string instead of a TupleTag.
   *
   * <p>This method is simpler for cases when a typed tuple-tag is not needed to extract a
   * PCollection, for example when using schema transforms.
   */
  public static <K, InputT> KeyedPCollectionTuple<K> of(String tag, PCollection<KV<K, InputT>> pc) {
    return of(new TupleTag<>(tag), pc);
  }

  /**
   * Returns a new {@code KeyedPCollectionTuple<K>} that is the same as this, appended with the
   * given PCollection.
   */
  public <V> KeyedPCollectionTuple<K> and(TupleTag<V> tag, PCollection<KV<K, V>> pc) {
    if (pc.getPipeline() != getPipeline()) {
      throw new IllegalArgumentException("PCollections come from different Pipelines");
    }
    TaggedKeyedPCollection<K, ?> wrapper = new TaggedKeyedPCollection<>(tag, pc);
    Coder<K> myKeyCoder = keyCoder == null ? getKeyCoder(pc) : keyCoder;
    List<TaggedKeyedPCollection<K, ?>> newKeyedCollections = copyAddLast(keyedCollections, wrapper);
    return new KeyedPCollectionTuple<>(
        getPipeline(), newKeyedCollections, schema.getTupleTagList().and(tag), myKeyCoder);
  }

  /**
   * A version of {@link #and(String, PCollection)} that takes in a string instead of a TupleTag.
   *
   * <p>This method is simpler for cases when a typed tuple-tag is not needed to extract a
   * PCollection, for example when using schema transforms.
   */
  public <V> KeyedPCollectionTuple<K> and(String tag, PCollection<KV<K, V>> pc) {
    return and(new TupleTag<>(tag), pc);
  }

View on GitHub (pinned to 12126d8942)