{"record":{"id":"1bae350de4096b24","repo":"apache/beam","slug":"externalwithmetadata-transform-only-supports-keys-of-type","errorCode":null,"errorMessage":"ExternalWithMetadata transform only supports keys of type nullable(byte[])","messagePattern":"ExternalWithMetadata transform only supports keys of type nullable\\(byte\\[\\]\\)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java","lineNumber":2503,"sourceCode":"      super(\"KafkaIO.RowsWithMetadata\");\n      this.read = read;\n    }\n\n    static class Builder<K, V>\n        implements ExternalTransformBuilder<Read.External.Configuration, PBegin, PCollection<Row>> {\n\n      @Override\n      public PTransform<PBegin, PCollection<Row>> buildExternal(\n          Read.External.Configuration config) {\n        Read.Builder<K, V> readBuilder = new AutoValue_KafkaIO_Read.Builder<>();\n        Read.Builder.setupExternalBuilder(readBuilder, config);\n\n        Class<Deserializer<K>> keyDeserializer =\n            (Class<Deserializer<K>>) resolveClass(config.keyDeserializer);\n        Coder<K> keyCoder = Read.Builder.resolveCoder(keyDeserializer);\n        if (!(keyCoder instanceof NullableCoder\n            && keyCoder.getCoderArguments().get(0) instanceof ByteArrayCoder)) {\n          throw new RuntimeException(\n              \"ExternalWithMetadata transform only supports keys of type nullable(byte[])\");\n        }\n        Class<Deserializer<V>> valueDeserializer =\n            (Class<Deserializer<V>>) resolveClass(config.valueDeserializer);\n        Coder<V> valueCoder = Read.Builder.resolveCoder(valueDeserializer);\n        if (!(valueCoder instanceof NullableCoder\n            && valueCoder.getCoderArguments().get(0) instanceof ByteArrayCoder)) {\n          throw new RuntimeException(\n              \"ExternalWithMetadata transform only supports values of type nullable(byte[])\");\n        }\n\n        return readBuilder.build().externalWithMetadata();\n      }\n    }\n\n    public static <K, V> ByteArrayKafkaRecord toExternalKafkaRecord(KafkaRecord<K, V> kafkaRecord) {\n      List<KafkaHeader> headers =\n          (kafkaRecord.getHeaders() == null)","sourceCodeStart":2485,"sourceCodeEnd":2521,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L2485-L2521","documentation":"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.","triggerScenarios":"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.","commonSituations":"Specifying StringDeserializer for keys in a Python-cross-language Kafka read; defaults changed or explicitly set to non-byte deserializers.","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."],"exampleFix":"// before\nconfig.keyDeserializer = StringDeserializer.class.getName()\n// after\nconfig.keyDeserializer = \"org.apache.kafka.common.serialization.ByteArrayDeserializer\"","handlingStrategy":"validation","validationCode":"Coder<?> keyCoder = KafkaIO.Read.Builder.resolveCoder(keyDeserializerClass);\nif (!(keyCoder instanceof NullableCoder && keyCoder.getCoderArguments().get(0) instanceof ByteArrayCoder)) {\n  throw new IllegalArgumentException(\"External with-metadata read requires nullable(byte[]) keys\");\n}","typeGuard":"boolean isNullableBytes(Coder<?> c) {\n  return c instanceof NullableCoder\n      && !c.getCoderArguments().isEmpty()\n      && c.getCoderArguments().get(0) instanceof ByteArrayCoder;\n}","tryCatchPattern":"try {\n  PCollection<Row> rows = pipeline.apply(KafkaIO.readAllExternalWithMetadata(config));\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"only supports keys\")) {\n    // fall back to byte[] keys and convert downstream\n  }\n}","preventionTips":["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"],"tags":["java","kafka","cross-language","coder"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}