apache/beam · error · IllegalArgumentException

requires a deterministic key coder in order to use state…

Error message

%s requires a deterministic key coder in order to use state and timers, the reason is:%n %s

What it means

ParDo requires that when a DoFn uses state or timers, the input PCollection's key coder must be deterministic, since state is keyed and runners rely on consistent byte encoding of keys. If KvCoder.getKeyCoder().verifyDeterministic() fails, the pipeline cannot be validated and this exception is thrown with the non-determinism reason.

Solutions

Make the key coder deterministic: implement a custom deterministic Coder for the key class and set it via input.setCoder(KvCoder.of(deterministicKeyCoder, valueCoder))
Use a structurally ordered key type (e.g. String, Long) or serialize the key canonically (sorted fields, fixed-width encodings)
Restructure the pipeline to use a deterministic surrogate key

Example fix

// before
PCollection<KV<MyKey, V>> kv = ...; // MyKey uses default java serialization coder
// after
kv.setCoder(KvCoder.of(new DeterministicMyKeyCoder(), vCoder));

Defensive patterns

Strategy: validation

Validate before calling

if (input.getCoder() instanceof KvCoder) {
  try {
    ((KvCoder<?, ?>) input.getCoder()).getKeyCoder().verifyDeterministic();
  } catch (Coder.NonDeterministicException e) {
    throw new IllegalArgumentException("Key coder not deterministic: " + e.getMessage());
  }
}

Type guard

boolean hasDeterministicKeyCoder(PCollection<?> input) {
  return input.getCoder() instanceof KvCoder
      && ((KvCoder<?, ?>) input.getCoder()).getKeyCoder() instanceof DeterministicCoder;
}

Try / catch

try {
  kv.apply(ParDo.of(statefulFn));
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("deterministic key coder")) {
    kv.setCoder(KvCoder.of(new DeterministicKeyCoder(), valueCoder));
    kv.apply(ParDo.of(statefulFn));
  } else { throw e; }
}

Prevention

Run verifyDeterministic() on key coders in unit tests for stateful pipelines
Prefer String/Long/byte[] keys for stateful DoFns
Write explicit deterministic Coders for custom key classes instead of relying on defaults

When it happens

Trigger: Applying a ParDo whose DoFn declares @StateId/@TimerId (or TimerFamily) over an input PCollection whose key coder is non-deterministic, e.g. the default coder for a custom key class or a coder over double/Map/unordered types.

Common situations: Grouping/stateful aggregation keyed by custom POJOs using the default Java serialization coder; keys typed as double or nested maps; Avro/POJO coders that don't guarantee stable field order.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java:447

      } catch (IllegalAccessException e) {
        throw new RuntimeException(e);
      }
    }
  }

  private static void validateStateApplicableForInput(PCollection<?> input) {
    Coder<?> inputCoder = input.getCoder();
    checkArgument(
        inputCoder instanceof KvCoder,
        "%s requires its input to use %s in order to use state and timers.",
        ParDo.class.getSimpleName(),
        KvCoder.class.getSimpleName());

    KvCoder<?, ?> kvCoder = (KvCoder<?, ?>) inputCoder;
    try {
      kvCoder.getKeyCoder().verifyDeterministic();
    } catch (Coder.NonDeterministicException exc) {
      throw new IllegalArgumentException(
          String.format(
              "%s requires a deterministic key coder in order to use state and timers, the reason is:%n %s",
              ParDo.class.getSimpleName(), exc.getMessage()));
    }
  }

  private static void validateSideInputTypes(
      Map<String, PCollectionView<?>> sideInputs, DoFn<?, ?> fn) {
    DoFnSignature signature = DoFnSignatures.getSignature(fn.getClass());
    DoFnSignature.ProcessElementMethod processElementMethod = signature.processElement();
    for (SideInputParameter sideInput : processElementMethod.getSideInputParameters()) {
      PCollectionView<?> view = sideInputs.get(sideInput.sideInputId());
      checkArgument(
          view != null,
          "the ProcessElement method expects a side input identified with the tag %s, but no such side input was"
              + " supplied. Use withSideInput(String, PCollectionView) to supply this side input.",
          sideInput.sideInputId());
      TypeDescriptor<?> viewType = view.getViewFn().getTypeDescriptor();

View on GitHub (pinned to 12126d8942)