{"record":{"id":"03dba8a47f89341c","repo":"apache/beam","slug":"could-not-decode-avro-record-from-given-bytes-bytes","errorCode":null,"errorMessage":"Could not decode avro record from given bytes ${bytes}","messagePattern":"Could not decode avro record from given bytes (.+?)","errorType":"exception","errorClass":"AvroRuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java","lineNumber":726,"sourceCode":"\n  private static class AvroBytesToRowFn extends SimpleFunction<byte[], Row> {\n    private final AvroCoder<GenericRecord> coder;\n    private final Schema beamSchema;\n\n    AvroBytesToRowFn(Schema beamSchema) {\n      org.apache.avro.Schema avroSchema = toAvroSchema(beamSchema);\n      coder = AvroCoder.of(avroSchema);\n      this.beamSchema = beamSchema;\n    }\n\n    @Override\n    public Row apply(byte[] bytes) {\n      try {\n        ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);\n        GenericRecord record = coder.decode(inputStream);\n        return AvroUtils.toBeamRowStrict(record, beamSchema);\n      } catch (Exception e) {\n        throw new AvroRuntimeException(\n            \"Could not decode avro record from given bytes \"\n                + new String(bytes, StandardCharsets.UTF_8),\n            e);\n      }\n    }\n  }\n\n  /** Returns a function mapping Beam {@link Row}s to encoded AVRO {@link GenericRecord}s. */\n  public static SimpleFunction<Row, byte[]> getRowToAvroBytesFunction(Schema beamSchema) {\n    return new RowToAvroBytesFn(beamSchema);\n  }\n\n  private static class RowToAvroBytesFn extends SimpleFunction<Row, byte[]> {\n    private final org.apache.avro.Schema avroSchema;\n    private final AvroCoder<GenericRecord> coder;\n\n    RowToAvroBytesFn(Schema beamSchema) {\n      avroSchema = toAvroSchema(beamSchema);","sourceCodeStart":708,"sourceCodeEnd":744,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java#L708-L744","documentation":"AvroUtils.getAvroBytesToRowFunction returns a SerializableFunction whose apply(byte[]) decodes bytes into an Avro GenericRecord and converts it to a Beam Row. Any exception during decoding (bad binary encoding, truncated/corrupt data, wrong writer schema for the coder) or during strict row conversion is rethrown as an AvroRuntimeException wrapping the raw bytes as a UTF-8 string, to make the offending payload visible.","triggerScenarios":"apply() invoked with byte arrays that are not valid Avro GenericRecord encodings for the configured coder, or decodable records that fail AvroUtils.toBeamRowStrict (e.g. schema/field mismatch), within the try block.","commonSituations":"Kafka/IO records whose writer schema changed after the coder/schema was fixed; concatenation of multiple messages into one byte array; records serialized with a different codec or container format (Avro data files vs raw records); null or empty payloads from upstream sources.","solutions":["Verify the writer schema of the incoming bytes matches the schema given to the coder; update the beamSchema/avroSchema if the producer evolved.","Inspect the payload echoed in the message for corruption, truncation, or a leading schema-version byte that must be stripped.","Ensure the source emits raw Avro binary records, not Avro Object Container Files (use appropriate reader for files).","Add input validation/filtering (or a dead-letter path) so malformed records are handled instead of failing the pipeline."],"exampleFix":"// before: blind decode\nRow row = bytesToRow.apply(bytes);\n// after: guard against empty/malformed input\nif (bytes == null || bytes.length == 0) { sendToDeadLetter(bytes); }\nelse { try { row = bytesToRow.apply(bytes); } catch (AvroRuntimeException e) { log.warn(\"bad record\", e); sendToDeadLetter(bytes); } }","handlingStrategy":"try-catch","validationCode":"if (bytes == null || bytes.length == 0) {\n  return deadLetter(bytes, \"empty payload\");\n}\n","typeGuard":null,"tryCatchPattern":"try {\n  return AvroUtils.getAvroBytesToRowFunction(beamSchema).apply(bytes);\n} catch (AvroRuntimeException e) {\n  log.warn(\"Unparseable avro record: {}\", e.getMessage());\n  return deadLetter(bytes, e); // or use a fallback schema / skip element\n}","preventionTips":["Pin the writer schema: use confluent wire-format or include schema version so decode always uses the right schema.","Validate payloads are raw Avro binary records (not container files) at the source.","Add dead-letter queues for malformed records instead of failing the whole pipeline.","Test decode round-trips whenever producer schema changes."],"tags":["avro","beam","deserialization","java"],"backgroundTag":"avro-decode-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}