apache/beam · error · RuntimeException

Unable to infer coder for KeyT

Error message

Unable to infer coder for KeyT (%s). Specify it explicitly using withOutputKeyCoder().

What it means

Watch.Growth must code the keys output by the output key function. If no key coder was provided with withOutputKeyCoder(), Beam infers KeyT from the function's type descriptor and asks the CoderRegistry; on CannotProvideCoderException it throws a RuntimeException instructing explicit specification. Common when KeyT is erased via a lambda or is an unregistered custom type.

Solutions

1. Call .withOutputKeyCoder(coder) with an explicit Coder<KeyT>
2. Use a concrete named class (not a raw lambda) for the key function so KeyT resolves
3. Register the key type's coder via CoderRegistry.registerCoderForType
4. Switch to a standard key type (String, Long) with a built-in coder

Example fix

// before
Watch.growthOf(pollFn).withOutputKeyFn(fn); // KeyT uninferrable
// after
Watch.growthOf(pollFn).withOutputKeyFn(fn).withOutputKeyCoder(StringUtf8Coder.of());

try {
  pipeline.getCoderRegistry().getCoder(TypeDescriptors.outputOf(keyFn));
} catch (CannotProvideCoderException e) {
  growth = growth.withOutputKeyCoder(StringUtf8Coder.of());
}

try { pc = input.apply(growth); } catch (RuntimeException e) { if (e.getMessage().contains("Unable to infer coder for KeyT")) { /* reapply with withOutputKeyCoder */ } }

When it happens

Trigger: Applying Watch.growthOf(...).withOutputKeyFn(fn) where fn's KeyT type parameter cannot be resolved to a registered coder — lambda/anonymous function erasure, or a custom key type with no coder in the registry.

Common situations: Passing lambdas for output key functions so type descriptors are incomplete; custom key classes (e.g. protobuf types or POJOs) not registered with the CoderRegistry.

Related errors

• Unable to infer coder for OutputT
• Key coder must be deterministic
• bad coder kind
• bad decoding function
• bad encoding function

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

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Watch.java:810

                  + "). Specify it explicitly using withOutputCoder().");
        }
      }

      Coder<KeyT> outputKeyCoder = getOutputKeyCoder();
      SerializableFunction<OutputT, KeyT> outputKeyFn = getOutputKeyFn();
      if (getOutputKeyFn() == null) {
        // This by construction can happen only if OutputT == KeyT
        outputKeyCoder = (Coder) outputCoder;
        outputKeyFn = (SerializableFunction) SerializableFunctions.identity();
      } else {
        if (outputKeyCoder == null) {
          // If a coder was not specified explicitly, infer it from the OutputT type parameter
          // of the output key fn.
          TypeDescriptor<KeyT> keyT = TypeDescriptors.outputOf(getOutputKeyFn());
          try {
            outputKeyCoder = input.getPipeline().getCoderRegistry().getCoder(keyT);
          } catch (CannotProvideCoderException e) {
            throw new RuntimeException(
                "Unable to infer coder for KeyT ("
                    + keyT
                    + "). Specify it explicitly using withOutputKeyCoder().");
          }
        }
        try {
          outputKeyCoder.verifyDeterministic();
        } catch (Coder.NonDeterministicException e) {
          throw new IllegalArgumentException(
              "Key coder " + outputKeyCoder + " must be deterministic");
        }
      }

      PCollection<KV<InputT, List<TimestampedValue<OutputT>>>> polledPc =
          input
              .apply(
                  ParDo.of(new WatchGrowthFn<>(this, outputCoder, outputKeyFn, outputKeyCoder))
                      .withSideInputs(getPollFn().getRequirements().getSideInputs()))

View on GitHub (pinned to 12126d8942)