apache/beam · error · IllegalArgumentException

PCollection does not use a KvCoder

Error message

PCollection does not use a KvCoder

What it means

CoGroupByKey needs the value coder of each input to build the union coder, and it assumes every input PCollection's coder is a KvCoder. If a PCollection's coder is not a KvCoder instance, the assumption is violated and an IllegalArgumentException is thrown.

Source

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

    PCollection<KV<K, CoGbkResult>> result =
        groupedTable.apply(
            "ConstructCoGbkResultFn", ParDo.of(new ConstructCoGbkResultFn<>(tupleTags)));
    result.setCoder(KvCoder.of(keyCoder, CoGbkResultCoder.of(tupleTags, unionCoder)));

    return result;
  }

  //////////////////////////////////////////////////////////////////////////////

  /**
   * Returns the value coder for the given PCollection. Assumes that the value coder is an instance
   * of {@code KvCoder<K, V>}.
   */
  private <V> Coder<V> getValueCoder(PCollection<KV<K, V>> pCollection) {
    // Assumes that the PCollection uses a KvCoder.
    Coder<?> entryCoder = pCollection.getCoder();
    if (!(entryCoder instanceof KvCoder<?, ?>)) {
      throw new IllegalArgumentException("PCollection does not use a KvCoder");
    }
    @SuppressWarnings("unchecked")
    KvCoder<K, V> coder = (KvCoder<K, V>) entryCoder;
    return coder.getValueCoder();
  }

  /**
   * Returns a UnionTable for the given input PCollection, using the given union index and the given
   * unionTableEncoder.
   */
  private <V> PCollection<KV<K, RawUnionValue>> makeUnionTable(
      final int index,
      PCollection<KV<K, V>> pCollection,
      KvCoder<K, RawUnionValue> unionTableEncoder) {

    return pCollection
        .apply("MakeUnionTable" + index, ParDo.of(new ConstructUnionTableFn<>(index)))
        .setCoder(unionTableEncoder);

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure each input PCollection has a KvCoder via KvCoder.of(keyCoder, valueCoder)
2. Call pc.setCoder(KvCoder.of(...)) explicitly on inputs before the join
3. Verify coder inference upstream — the transform producing the KV should set a KvCoder

Example fix

// before
pc.apply(CoGroupByKey.create()); // coder not KvCoder
// after
pc.setCoder(KvCoder.of(keyCoder, valueCoder));
pc.apply(CoGroupByKey.create());

if (!(pc.getCoder() instanceof KvCoder)) { throw new IllegalArgumentException("Input must have a KvCoder"); }

Coder<?> c = pc.getCoder();
if (c instanceof KvCoder<?, ?> kvCoder) { /* use kvCoder */ }

try { coder = getValueCoder(pc); } catch (IllegalArgumentException e) { pc.setCoder(KvCoder.of(defaultKeyCoder, defaultValueCoder)); }

When it happens

Trigger: Passing a PCollection<KV<K,V>> whose coder was not inferred as KvCoder, e.g. after setCoder() with a non-KvCoder or a custom source that produced an unspecified coder.

Common situations: Custom I/O or PTransforms that set explicit wrong coders; using coders.setCoder with the wrong type; older Beam versions where coder inference differed.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• PCollection does not use a KvCoder
• Invalid encoded string length: {}
• cannot encode a null String
• cannot encode a null Integer
• error when decoding a textual integer

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