apache/beam · error · IllegalStateException

GroupByKey requires its input to use KvCoder

Error message

GroupByKey requires its input to use KvCoder

What it means

GroupByKey's coder handling assumes the input PCollection's coder is a KvCoder<K,V> so it can extract key/value coders. getInputKvCoder() type-checks the coder and throws IllegalStateException if it is any other Coder, since grouping is meaningless without KV-shaped coders.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java:298

    // This primitive operation groups by the combination of key and window,
    // merging windows as needed, using the windows assigned to the
    // key/value input elements and the window merge operation of the
    // window function associated with the input PCollection.
    return PCollection.createPrimitiveOutputInternal(
        input.getPipeline(),
        updateWindowingStrategy(input.getWindowingStrategy()),
        input.isBounded(),
        getOutputKvCoder(input.getCoder()));
  }

  /**
   * Returns the {@code Coder} of the input to this transform, which should be a {@code KvCoder}.
   */
  @SuppressWarnings("unchecked")
  static <K, V> KvCoder<K, V> getInputKvCoder(Coder<?> inputCoder) {
    if (!(inputCoder instanceof KvCoder)) {
      throw new IllegalStateException("GroupByKey requires its input to use KvCoder");
    }
    return (KvCoder<K, V>) inputCoder;
  }

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

  /**
   * Returns the {@code Coder} of the keys of the input to this transform, which is also used as the
   * {@code Coder} of the keys of the output of this transform.
   */
  public static <K, V> Coder<K> getKeyCoder(Coder<KV<K, V>> inputCoder) {
    return GroupByKey.<K, V>getInputKvCoder(inputCoder).getKeyCoder();
  }

  /** Returns the {@code Coder} of the values of the input to this transform. */
  public static <K, V> Coder<V> getInputValueCoder(Coder<KV<K, V>> inputCoder) {
    return GroupByKey.<K, V>getInputKvCoder(inputCoder).getValueCoder();
  }

View on GitHub (pinned to 12126d8942)

Solutions

1. Remove the erroneous setCoder() call so Beam infers KvCoder from the KV generics
2. Explicitly set the correct coder: input.setCoder(KvCoder.of(keyCoder, valueCoder))
3. If the element type genuinely isn't KV<K,V>, add a MapElements step producing KV before GroupByKey
4. Check upstream transforms (e.g. custom sources) for coder overrides that replaced the KvCoder

Example fix

// before
kvCollection.setCoder(SerializableCoder.of(KV.class));
kvCollection.apply(GroupByKey.create());
// after
kvCollection.setCoder(KvCoder.of(StringUtf8Coder.of(), valueCoder));
kvCollection.apply(GroupByKey.create());

if (!(input.getCoder() instanceof KvCoder)) {
  input.setCoder(KvCoder.of(keyCoder, valueCoder));
}

static <K,V> boolean hasKvCoder(PCollection<KV<K,V>> pc) {
  return pc.getCoder() instanceof KvCoder<?, ?>;
}

try {
  grouped = input.apply(GroupByKey.create());
} catch (IllegalStateException e) {
  if (e.getMessage().contains("requires its input to use KvCoder")) { /* fix input coder */ }
  else throw e;
}

When it happens

Trigger: Applying GroupByKey to a PCollection whose coder was set (or inferred) to something other than KvCoder — e.g. input.setCoder(SomeCustomCoder) on a KV-typed collection, an in-memory/create() source with a mismatched coder, or an intermediate transform that replaced the coder.

Common situations: Overriding coders on KV collections for debugging; using PCollectionList/PBegin glue that lost the KvCoder; consuming from a source that set a generic SerializableCoder for KV values; grouping after a transform that returned a different logical type but kept KV generics.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• the GroupByKey requires its output coder to be %s but found
• the keyCoder of a GroupByKey must be deterministic
• SortValues requires its input to use KvCoder
• SortValues requires the values be encoded with IterableCoder
• cannot encode a null Integer

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