apache/beam · error · RuntimeException

ExternalWithMetadata transform only supports values of type…

Error message

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

What it means

Same restriction as for keys, applied to the value deserializer: the external-with-metadata KafkaIO transform only supports value coders of nullable(byte[]), since records are shipped as raw bytes plus metadata across the language boundary.

Solutions

Set valueDeserializer to org.apache.kafka.common.serialization.ByteArrayDeserializer.
Decode Avro/JSON payloads downstream using ParseAvro/ParseJson or a SchemaRegistry transform after the read.
Use plain Java KafkaIO.read() with explicit deserializers/coders when off the cross-language path.

Example fix

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

Defensive patterns

Strategy: validation

Validate before calling

Coder<?> valueCoder = KafkaIO.Read.Builder.resolveCoder(valueDeserializerClass);
if (!(valueCoder instanceof NullableCoder && valueCoder.getCoderArguments().get(0) instanceof ByteArrayCoder)) {
  throw new IllegalArgumentException("External with-metadata read requires nullable(byte[]) values; parse Avro/JSON downstream");
}

Type guard

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

Try / catch

try {
  pipeline.apply(KafkaIO.readAllExternalWithMetadata(config));
} catch (RuntimeException e) {
  if (e.getMessage() != null && e.getMessage().contains("only supports values")) {
    config.setValueDeserializer("org.apache.kafka.common.serialization.ByteArrayDeserializer");
  }
}

Prevention

Never set Avro/Confluent deserializers in external-with-metadata configs
Do payload decoding (ParseAvro/SchemaRegistry transforms) after the read
Add a config sanity test asserting ByteArrayDeserializer for both key and value

When it happens

Trigger: Expanding KafkaIO externalWithMetadata with config.valueDeserializer that resolves to anything other than NullableCoder(ByteArrayCoder) — e.g. KafkaAvroDeserializer or StringDeserializer for values.

Common situations: Using Avro/Confluent deserializer in the cross-language read (should be handled after read via schema registry in the consuming transform), or StringDeserializer for values.

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/4987d93841890044. Report an issue: GitHub.

Appendix: source

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

      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)
              ? null
              : Arrays.stream(kafkaRecord.getHeaders().toArray())
                  .map(h -> new KafkaHeader(h.key(), h.value()))
                  .collect(Collectors.toList());
      ByteArrayKafkaRecord byteArrayKafkaRecord =
          new ByteArrayKafkaRecord(
              kafkaRecord.getTopic(),
              kafkaRecord.getPartition(),

View on GitHub (pinned to 12126d8942)