{"record":{"id":"410dd6d69202ca85","repo":"apache/beam","slug":"expected-input-coder-to-be-kvcoder-but-was","errorCode":null,"errorMessage":"Expected input coder to be KvCoder, but was ","messagePattern":"Expected input coder to be KvCoder, but was ","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java","lineNumber":1691,"sourceCode":"    @Override\n    protected String getKindString() {\n      return String.format(\"Combine.perKeyWithFanout(%s)\", NameUtils.approximateSimpleName(fn));\n    }\n\n    @Override\n    public PCollection<KV<K, OutputT>> expand(PCollection<KV<K, InputT>> input) {\n      return applyHelper(input);\n    }\n\n    private <AccumT> PCollection<KV<K, OutputT>> applyHelper(PCollection<KV<K, InputT>> input) {\n\n      // Name the accumulator type.\n      @SuppressWarnings(\"unchecked\")\n      final GlobalCombineFn<InputT, AccumT, OutputT> typedFn =\n          (GlobalCombineFn<InputT, AccumT, OutputT>) this.fn;\n\n      if (!(input.getCoder() instanceof KvCoder)) {\n        throw new IllegalStateException(\n            \"Expected input coder to be KvCoder, but was \" + input.getCoder());\n      }\n\n      @SuppressWarnings(\"unchecked\")\n      final KvCoder<K, InputT> inputCoder = (KvCoder<K, InputT>) input.getCoder();\n      final Coder<AccumT> accumCoder;\n\n      try {\n        accumCoder =\n            typedFn.getAccumulatorCoder(\n                input.getPipeline().getCoderRegistry(), inputCoder.getValueCoder());\n      } catch (CannotProvideCoderException e) {\n        throw new IllegalStateException(\"Unable to determine accumulator coder.\", e);\n      }\n      Coder<InputOrAccum<InputT, AccumT>> inputOrAccumCoder =\n          new InputOrAccum.InputOrAccumCoder<>(inputCoder.getValueCoder(), accumCoder);\n\n      // A CombineFn's mergeAccumulator can be applied in a tree-like fashion.","sourceCodeStart":1673,"sourceCodeEnd":1709,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java#L1673-L1709","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the input coder is KvCoder.of(keyCoder, valueCoder) via setCoder","Let Beam infer coders (remove manual setCoder overrides)","Check upstream transforms that replaced the coder with a wrapper coder"],"exampleFix":"// before\nkvPCollection.setCoder(myCustomCoder);\nkvPCollection.apply(Combine.perKey(fn));\n// after\nkvPCollection.setCoder(KvCoder.of(keyCoder, valueCoder));\nkvPCollection.apply(Combine.perKey(fn));","handlingStrategy":"type-guard","validationCode":"if (!(pc.getCoder() instanceof KvCoder)) {\n  throw new IllegalStateException(\"Input to Combine.perKey must have KvCoder, got \" + pc.getCoder());\n}","typeGuard":"boolean hasKvCoder(PCollection<?> pc) {\n  return pc.getCoder() instanceof KvCoder;\n}","tryCatchPattern":"try {\n  return pc.apply(Combine.perKey(fn));\n} catch (IllegalStateException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"KvCoder\")) {\n    pc.setCoder(KvCoder.of(keyCoder, valueCoder));\n    return pc.apply(Combine.perKey(fn));\n  }\n  throw e;\n}","preventionTips":["Don't override coders on KV collections with non-KvCoder types","Rely on Beam's coder inference unless necessary","Check upstream transforms that might wrap or replace coders"],"tags":["java","coder","combine"],"backgroundTag":"incompatible-source-type","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}