{"record":{"id":"b5d365ab026bdc51","repo":"apache/beam","slug":"unable-to-determine-accumulator-coder","errorCode":null,"errorMessage":"Unable to determine accumulator coder.","messagePattern":"Unable to determine accumulator coder\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java","lineNumber":1704,"sourceCode":"      @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.\n      // Here we shard the key using an integer nonce, combine on that partial\n      // set of values, then drop the nonce and do a final combine of the\n      // aggregates.  We do this by splitting the original CombineFn into two,\n      // on that does addInput + merge and another that does merge + extract.\n      GlobalCombineFn<InputT, AccumT, AccumT> hotPreCombine;\n      GlobalCombineFn<InputOrAccum<InputT, AccumT>, AccumT, OutputT> postCombine;\n      if (typedFn instanceof CombineFn) {\n        final CombineFn<InputT, AccumT, OutputT> fn = (CombineFn<InputT, AccumT, OutputT>) typedFn;\n        hotPreCombine =\n            new CombineFn<InputT, AccumT, AccumT>() {\n              @Override\n              public AccumT createAccumulator() {\n                return fn.createAccumulator();","sourceCodeStart":1686,"sourceCodeEnd":1722,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java#L1686-L1722","documentation":"During expand, Combine.PerKey asks the GlobalCombineFn for an accumulator coder via the pipeline's CoderRegistry; if the fn cannot provide one (CannotProvideCoderException) an IllegalStateException is thrown because the shuffle stages cannot be serialized.","triggerScenarios":"A custom CombineFn whose getAccumulatorCoder throws CannotProvideCoderException (e.g. accumulator type is generic/unresolvable or getAccumulatorCoder isn't properly implemented), then expand().","commonSituations":"Custom CombineFns with non-serializable or generic accumulator classes; anonymous inner classes losing type info; using a fn that relies on defaults not registered in the CoderRegistry.","solutions":["Implement getAccumulatorCoder to return a concrete Coder for your AccumT","Register a custom coder in the pipeline's CoderRegistry","Make AccumT a concrete, top-level serializable class","Use a built-in CombineFn (Sum, Count, etc.) whose coders are known"],"exampleFix":"// before\nclass MyFn extends CombineFn<In, Map<String,Object>, Out> {} // uncodable accum\n// after\nclass MyAccum { List<String> values; }\nclass MyFn extends CombineFn<In, MyAccum, Out> {\n  public Coder<MyAccum> getAccumulatorCoder(CoderRegistry r, Coder<In> in) {\n    return SerializableCoder.of(MyAccum.class);\n  }\n}","handlingStrategy":"validation","validationCode":"try {\n  fn.getAccumulatorCoder(CoderRegistry.createDefault(), inputCoder);\n} catch (CannotProvideCoderException e) {\n  throw new IllegalStateException(\"CombineFn cannot provide accumulator coder; fix getAccumulatorCoder\", e);\n}","typeGuard":null,"tryCatchPattern":"try {\n  return pc.apply(Combine.perKey(fn));\n} catch (IllegalStateException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"accumulator coder\")) {\n    LOG.error(\"Register a coder for AccumT in CoderRegistry or implement getAccumulatorCoder\", e);\n  }\n  throw e;\n}","preventionTips":["Always implement getAccumulatorCoder in custom CombineFns","Use concrete, top-level, serializable Accum classes (no anonymous/generic accumulators)","Register custom coders in the pipeline CoderRegistry","Unit-test combine fns with a default CoderRegistry before pipeline runs"],"tags":["java","coder","combine"],"backgroundTag":"coder-inference-failed","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"}