apache/beam · error · RuntimeException

Couldn't resolve coder for Deserializer:

Error message

Couldn't resolve coder for Deserializer:

What it means

KafkaIO's external-transform builder calls resolveCoder to derive a Coder for a given key/value Deserializer class. It only recognizes a small set of well-known deserializers (ByteArray, Long, Integer, String, Avro/Protobuf via SchemaRegistry); for anything else it cannot infer a Coder and throws. This happens when the KafkaIO read is exported to a cross-language pipeline and the configured deserializer is not one of the supported types.

Solutions

  1. Use a supported deserializer: ByteArrayDeserializer, StringDeserializer, LongSerializer/Integer variants, or KafkaAvroDeserializer/ProtobufConfluentByteUtils-based ones.
  2. Wrap byte-array style deserializers so the inferred coder is NullableCoder(ByteArrayCoder) where required.
  3. If a custom deserializer is required, don't use the external/cross-language path; build the read directly in Java with explicit .withKeyCoder/.withValueCoder.
  4. Verify the deserializer class name string resolves to the intended class (a wrong name yields the generic failure).

Example fix

// before
.withValueDeserializerAndCoder(MyCustomDeserializer.class, myCoder) // external path
// after
.withValueDeserializer(ByteArrayDeserializer.class) // resolvable by external builder
Defensive patterns

Strategy: validation

Validate before calling

Set<String> supported = Set.of("ByteArrayDeserializer","StringDeserializer","LongDeserializer","IntegerDeserializer","ByteBufferDeserializer","KafkaAvroDeserializer");
if (!supported.stream().anyMatch(d -> deserializerCls.getName().contains(d))) {
  throw new IllegalArgumentException("External KafkaIO read cannot infer coder for " + deserializerCls.getName());
}

Type guard

boolean isExternallyResolvable(Class<?> d) {
  return Arrays.stream(KafkaIO.Read.Builder.resolveCoder(d).getClass().getDeclaredFields()).count() >= 0
      && (Coder.class.isAssignableFrom(d.getClass()) || SupportedDeserializers.ALL.contains(d.getName()));
}

Prevention

When it happens

Trigger: Calling setupExternalBuilder / expanding an external KafkaIO.Read (e.g. via SchemaTransform or Python xlang) with a keyDeserializer or valueDeserializer class not in resolveCoder's known list, or a custom org.apache.kafka.common.serialization.Deserializer implementation.

Common situations: Using custom deserializers (e.g. company-specific JSON deserializer), or deserializers like LongSerializer without nullable wrapper in external/cross-language mode; also typos in the deserializer class name.

Related errors

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

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java:1006

      private static <T> Coder<T> resolveCoder(Class<Deserializer<T>> deserializer) {
        for (Method method : deserializer.getDeclaredMethods()) {
          if (method.getName().equals("deserialize")) {
            Class<?> returnType = method.getReturnType();
            if (returnType.equals(Object.class)) {
              continue;
            }
            if (returnType.equals(byte[].class)) {
              return (Coder<T>) NullableCoder.of(ByteArrayCoder.of());
            } else if (returnType.equals(Integer.class)) {
              return (Coder<T>) NullableCoder.of(VarIntCoder.of());
            } else if (returnType.equals(Long.class)) {
              return (Coder<T>) NullableCoder.of(VarLongCoder.of());
            } else {
              throw new RuntimeException("Couldn't infer Coder from " + deserializer);
            }
          }
        }
        throw new RuntimeException("Couldn't resolve coder for Deserializer: " + deserializer);
      }
    }

    /**
     * Exposes {@link KafkaIO.TypedWithoutMetadata} as an external transform for cross-language
     * usage.
     */
    @AutoService(ExternalTransformRegistrar.class)
    public static class External implements ExternalTransformRegistrar {

      // Using the transform name in the URN so that the corresponding transform can be easily
      // identified.
      public static final String URN_WITH_METADATA =
          "beam:transform:org.apache.beam:kafka_read_with_metadata:v1";
      public static final String URN_WITHOUT_METADATA =
          "beam:transform:org.apache.beam:kafka_read_without_metadata:v1";

      @Override

View on GitHub (pinned to 12126d8942)