{"record":{"id":"5e82104c11485db6","repo":"apache/beam","slug":"could-not-determine-coder-for-accumulator","errorCode":null,"errorMessage":"Could not determine coder for accumulator","messagePattern":"Could not determine coder for accumulator","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/util/AppliedCombineFn.java","lineNumber":114,"sourceCode":"  }\n\n  public static <K, InputT, AccumT, OutputT>\n      AppliedCombineFn<K, InputT, AccumT, OutputT> withInputCoder(\n          GlobalCombineFn<? super InputT, AccumT, OutputT> fn,\n          CoderRegistry registry,\n          KvCoder<K, InputT> kvCoder,\n          Iterable<PCollectionView<?>> sideInputViews,\n          WindowingStrategy<?, ?> windowingStrategy) {\n    // Casting down the K and InputT is safe because they're only used as inputs.\n    @SuppressWarnings(\"unchecked\")\n    GlobalCombineFn<InputT, AccumT, OutputT> clonedFn =\n        (GlobalCombineFn<InputT, AccumT, OutputT>) SerializableUtils.clone(fn);\n    try {\n      Coder<AccumT> accumulatorCoder =\n          clonedFn.getAccumulatorCoder(registry, kvCoder.getValueCoder());\n      return create(clonedFn, accumulatorCoder, sideInputViews, kvCoder, windowingStrategy);\n    } catch (CannotProvideCoderException e) {\n      throw new IllegalStateException(\"Could not determine coder for accumulator\", e);\n    }\n  }\n\n  private static <K, InputT, AccumT, OutputT> AppliedCombineFn<K, InputT, AccumT, OutputT> create(\n      GlobalCombineFn<InputT, AccumT, OutputT> fn,\n      Coder<AccumT> accumulatorCoder,\n      Iterable<PCollectionView<?>> sideInputViews,\n      KvCoder<K, InputT> kvCoder,\n      WindowingStrategy<?, ?> windowingStrategy) {\n    return new AppliedCombineFn<>(fn, accumulatorCoder, sideInputViews, kvCoder, windowingStrategy);\n  }\n\n  public GlobalCombineFn<InputT, AccumT, OutputT> getFn() {\n    return fn;\n  }\n\n  public Iterable<PCollectionView<?>> getSideInputViews() {\n    return sideInputViews;","sourceCodeStart":96,"sourceCodeEnd":132,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/util/AppliedCombineFn.java#L96-L132","documentation":"AppliedCombineFn.withInputCoder() clones the given GlobalCombineFn and asks it to infer a Coder for its accumulator via getAccumulatorCoder(registry, kvCoder.getValueCoder()). If the combine function cannot provide a coder (CannotProvideCoderException), the factory wraps it in an IllegalStateException('Could not determine coder for accumulator'). The CombineFn must be able to serialize its accumulator type for distributed execution.","triggerScenarios":"Creating an AppliedCombineFn for a custom CombineFn whose getAccumulatorCoder throws CannotProvideCoderException — typically because the accumulator class has no registered default coder and the type is not one Beam can infer (e.g. a custom POJO without an encoded schema).","commonSituations":"Writing a custom CombineFn with an accumulator type like Optional, an interface, or a POJO lacking a registered Coder; registry in withInputCoder(CoderRegistry, ...) not containing coders for nested accumulator types.","solutions":["Implement getAccumulatorCoder() in your CombineFn to return an explicit Coder (e.g. a SerializableCoder or custom Coder) instead of delegating/throwing.","Register a default Coder for the accumulator type in the CoderRegistry (registry.registerCoder(AccumT.class, MyCoder.class)).","Choose an accumulator type Beam can encode natively (KV, primitives, annotated with @DefaultSchema) so inference succeeds."],"exampleFix":"// before\n@Override public Coder<Accum> getAccumulatorCoder(CoderRegistry r, Coder<Input> in) throws CannotProvideCoderException {\n  throw new CannotProvideCoderException(\"no coder\");\n}\n// after\n@Override public Coder<Accum> getAccumulatorCoder(CoderRegistry r, Coder<Input> in) {\n  return SerializableCoder.of(Accum.class);\n}","handlingStrategy":"validation","validationCode":"CoderRegistry registry = pipeline.getCoderRegistry();\nregistry.registerCoderForClass(MyAccum.class, MyAccumCoder.class);\n// or verify your CombineFn.getAccumulatorCoder does not throw CannotProvideCoderException","typeGuard":null,"tryCatchPattern":"try {\n  AppliedCombineFn.withInputCoder(fn, registry, kvCoder, sideInputs);\n} catch (IllegalStateException e) {\n  // fall back to a coder-explicit CombineFn implementation\n}","preventionTips":["Always implement getAccumulatorCoder() explicitly in custom CombineFns.","Register default coders for accumulator types in the CoderRegistry.","Prefer accumulator types with natural encodings (primitives, KV, schemas)."],"tags":["java","apache-beam","coder","combinefn","serialization"],"backgroundTag":"type-mismatch","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"}