apache/beam · error · RuntimeException

ExternalWithMetadata transform only supports keys of type…

Error message

ExternalWithMetadata transform only supports keys of type nullable(byte[])

What it means

The external (cross-language) KafkaIO transform with metadata restricts the key type to NullableCoder wrapping ByteArrayCoder, because the external representation serializes raw record bytes. Any other resolved key coder is rejected at expansion time.

Solutions

Set keyDeserializer to org.apache.kafka.common.serialization.ByteArrayDeserializer.
If the pipeline expects String keys, read as byte[] and convert downstream with a MapElements step.
Use the non-external Java KafkaIO.read() with explicit coders if other key types are needed.

Example fix

// before
config.keyDeserializer = StringDeserializer.class.getName()
// after
config.keyDeserializer = "org.apache.kafka.common.serialization.ByteArrayDeserializer"

Defensive patterns

Strategy: validation

Validate before calling

Coder<?> keyCoder = KafkaIO.Read.Builder.resolveCoder(keyDeserializerClass);
if (!(keyCoder instanceof NullableCoder && keyCoder.getCoderArguments().get(0) instanceof ByteArrayCoder)) {
  throw new IllegalArgumentException("External with-metadata read requires nullable(byte[]) keys");
}

Type guard

boolean isNullableBytes(Coder<?> c) {
  return c instanceof NullableCoder
      && !c.getCoderArguments().isEmpty()
      && c.getCoderArguments().get(0) instanceof ByteArrayCoder;
}

Try / catch

try {
  PCollection<Row> rows = pipeline.apply(KafkaIO.readAllExternalWithMetadata(config));
} catch (RuntimeException e) {
  if (e.getMessage() != null && e.getMessage().contains("only supports keys")) {
    // fall back to byte[] keys and convert downstream
  }
}

Prevention

Default keyDeserializer to ByteArrayDeserializer in cross-language configs
Convert String/Integer keys after the read with MapElements
Document the nullable(byte[]) restriction wherever the external read config is built

When it happens

Trigger: Using KafkaIO externalWithMetadata / ReadAllExternalWithMetadata (or the Kafka SchemaTransform) where config.keyDeserializer resolves to a coder other than nullable(byte[]) — e.g. StringDeserializer or LongSerializer as key.

Common situations: Specifying StringDeserializer for keys in a Python-cross-language Kafka read; defaults changed or explicitly set to non-byte deserializers.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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

Appendix: source

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

      super("KafkaIO.RowsWithMetadata");
      this.read = read;
    }

    static class Builder<K, V>
        implements ExternalTransformBuilder<Read.External.Configuration, PBegin, PCollection<Row>> {

      @Override
      public PTransform<PBegin, PCollection<Row>> buildExternal(
          Read.External.Configuration config) {
        Read.Builder<K, V> readBuilder = new AutoValue_KafkaIO_Read.Builder<>();
        Read.Builder.setupExternalBuilder(readBuilder, config);

        Class<Deserializer<K>> keyDeserializer =
            (Class<Deserializer<K>>) resolveClass(config.keyDeserializer);
        Coder<K> keyCoder = Read.Builder.resolveCoder(keyDeserializer);
        if (!(keyCoder instanceof NullableCoder
            && keyCoder.getCoderArguments().get(0) instanceof ByteArrayCoder)) {
          throw new RuntimeException(
              "ExternalWithMetadata transform only supports keys of type nullable(byte[])");
        }
        Class<Deserializer<V>> valueDeserializer =
            (Class<Deserializer<V>>) resolveClass(config.valueDeserializer);
        Coder<V> valueCoder = Read.Builder.resolveCoder(valueDeserializer);
        if (!(valueCoder instanceof NullableCoder
            && valueCoder.getCoderArguments().get(0) instanceof ByteArrayCoder)) {
          throw new RuntimeException(
              "ExternalWithMetadata transform only supports values of type nullable(byte[])");
        }

        return readBuilder.build().externalWithMetadata();
      }
    }

    public static <K, V> ByteArrayKafkaRecord toExternalKafkaRecord(KafkaRecord<K, V> kafkaRecord) {
      List<KafkaHeader> headers =
          (kafkaRecord.getHeaders() == null)

View on GitHub (pinned to 12126d8942)