apache/beam · error · IllegalStateException

Unable to determine accumulator coder.

Error message

Unable to determine accumulator coder.

What it means

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.

Solutions

1. Implement getAccumulatorCoder to return a concrete Coder for your AccumT
2. Register a custom coder in the pipeline's CoderRegistry
3. Make AccumT a concrete, top-level serializable class
4. Use a built-in CombineFn (Sum, Count, etc.) whose coders are known

Example fix

// before
class MyFn extends CombineFn<In, Map<String,Object>, Out> {} // uncodable accum
// after
class MyAccum { List<String> values; }
class MyFn extends CombineFn<In, MyAccum, Out> {
  public Coder<MyAccum> getAccumulatorCoder(CoderRegistry r, Coder<In> in) {
    return SerializableCoder.of(MyAccum.class);
  }
}

try {
  fn.getAccumulatorCoder(CoderRegistry.createDefault(), inputCoder);
} catch (CannotProvideCoderException e) {
  throw new IllegalStateException("CombineFn cannot provide accumulator coder; fix getAccumulatorCoder", e);
}

try {
  return pc.apply(Combine.perKey(fn));
} catch (IllegalStateException e) {
  if (e.getMessage() != null && e.getMessage().contains("accumulator coder")) {
    LOG.error("Register a coder for AccumT in CoderRegistry or implement getAccumulatorCoder", e);
  }
  throw e;
}

When it happens

Trigger: A custom CombineFn whose getAccumulatorCoder throws CannotProvideCoderException (e.g. accumulator type is generic/unresolvable or getAccumulatorCoder isn't properly implemented), then expand().

Common situations: 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.

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
• Expected input coder to be KvCoder, but was
• ApproximateUnique.PerKey requires its input to use KvCoder

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

Appendix: source

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

      @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.
      // Here we shard the key using an integer nonce, combine on that partial
      // set of values, then drop the nonce and do a final combine of the
      // aggregates.  We do this by splitting the original CombineFn into two,
      // on that does addInput + merge and another that does merge + extract.
      GlobalCombineFn<InputT, AccumT, AccumT> hotPreCombine;
      GlobalCombineFn<InputOrAccum<InputT, AccumT>, AccumT, OutputT> postCombine;
      if (typedFn instanceof CombineFn) {
        final CombineFn<InputT, AccumT, OutputT> fn = (CombineFn<InputT, AccumT, OutputT>) typedFn;
        hotPreCombine =
            new CombineFn<InputT, AccumT, AccumT>() {
              @Override
              public AccumT createAccumulator() {
                return fn.createAccumulator();

View on GitHub (pinned to 12126d8942)