apache/beam · error · IllegalStateException

Expected input coder to be KvCoder, but was

Error message

Expected input coder to be KvCoder, but was 

What it means

Combine.PerKey requires its input PCollection's coder to be a KvCoder so key/value coders can be extracted for downstream stages; the expand() method throws IllegalStateException when the input coder is anything else.

Solutions

1. Ensure the input coder is KvCoder.of(keyCoder, valueCoder) via setCoder
2. Let Beam infer coders (remove manual setCoder overrides)
3. Check upstream transforms that replaced the coder with a wrapper coder

Example fix

// before
kvPCollection.setCoder(myCustomCoder);
kvPCollection.apply(Combine.perKey(fn));
// after
kvPCollection.setCoder(KvCoder.of(keyCoder, valueCoder));
kvPCollection.apply(Combine.perKey(fn));

if (!(pc.getCoder() instanceof KvCoder)) {
  throw new IllegalStateException("Input to Combine.perKey must have KvCoder, got " + pc.getCoder());
}

boolean hasKvCoder(PCollection<?> pc) {
  return pc.getCoder() instanceof KvCoder;
}

try {
  return pc.apply(Combine.perKey(fn));
} catch (IllegalStateException e) {
  if (e.getMessage() != null && e.getMessage().contains("KvCoder")) {
    pc.setCoder(KvCoder.of(keyCoder, valueCoder));
    return pc.apply(Combine.perKey(fn));
  }
  throw e;
}

When it happens

Trigger: Applying Combine.perKey() to a PCollection<KV<K,V>> whose coder was set explicitly to a non-KvCoder, or whose coder inference produced a raw/custom coder not extending KvCoder.

Common situations: Manually calling setCoder with a custom coder on a KV collection; a FileIO/source producing KVs with an inferred StructuredRecord-like coder; library versions where coder inference changed.

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

• Combine.GroupedValues requires its input to use KvCoder
• Combine.GroupedValues requires its input values to use…
• Could not obtain a Coder for the accumulator
• Unable to determine accumulator coder.
• ApproximateUnique.PerKey requires its input to use KvCoder

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

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java:1691

    @Override
    protected String getKindString() {
      return String.format("Combine.perKeyWithFanout(%s)", NameUtils.approximateSimpleName(fn));
    }

    @Override
    public PCollection<KV<K, OutputT>> expand(PCollection<KV<K, InputT>> input) {
      return applyHelper(input);
    }

    private <AccumT> PCollection<KV<K, OutputT>> applyHelper(PCollection<KV<K, InputT>> input) {

      // Name the accumulator type.
      @SuppressWarnings("unchecked")
      final GlobalCombineFn<InputT, AccumT, OutputT> typedFn =
          (GlobalCombineFn<InputT, AccumT, OutputT>) this.fn;

      if (!(input.getCoder() instanceof KvCoder)) {
        throw new IllegalStateException(
            "Expected input coder to be KvCoder, but was " + input.getCoder());
      }

      @SuppressWarnings("unchecked")
      final KvCoder<K, InputT> inputCoder = (KvCoder<K, InputT>) input.getCoder();
      final Coder<AccumT> accumCoder;

      try {
        accumCoder =
            typedFn.getAccumulatorCoder(
                input.getPipeline().getCoderRegistry(), inputCoder.getValueCoder());
      } catch (CannotProvideCoderException e) {
        throw new IllegalStateException("Unable to determine accumulator coder.", e);
      }
      Coder<InputOrAccum<InputT, AccumT>> inputOrAccumCoder =
          new InputOrAccum.InputOrAccumCoder<>(inputCoder.getValueCoder(), accumCoder);

      // A CombineFn's mergeAccumulator can be applied in a tree-like fashion.

View on GitHub (pinned to 12126d8942)