apache/beam · error · AvroRuntimeException

Could not decode avro record from given bytes ${bytes}

Error message

Could not decode avro record from given bytes ${bytes}

What it means

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.

Source

Thrown at sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java:726

  private static class AvroBytesToRowFn extends SimpleFunction<byte[], Row> {
    private final AvroCoder<GenericRecord> coder;
    private final Schema beamSchema;

    AvroBytesToRowFn(Schema beamSchema) {
      org.apache.avro.Schema avroSchema = toAvroSchema(beamSchema);
      coder = AvroCoder.of(avroSchema);
      this.beamSchema = beamSchema;
    }

    @Override
    public Row apply(byte[] bytes) {
      try {
        ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
        GenericRecord record = coder.decode(inputStream);
        return AvroUtils.toBeamRowStrict(record, beamSchema);
      } catch (Exception e) {
        throw new AvroRuntimeException(
            "Could not decode avro record from given bytes "
                + new String(bytes, StandardCharsets.UTF_8),
            e);
      }
    }
  }

  /** Returns a function mapping Beam {@link Row}s to encoded AVRO {@link GenericRecord}s. */
  public static SimpleFunction<Row, byte[]> getRowToAvroBytesFunction(Schema beamSchema) {
    return new RowToAvroBytesFn(beamSchema);
  }

  private static class RowToAvroBytesFn extends SimpleFunction<Row, byte[]> {
    private final org.apache.avro.Schema avroSchema;
    private final AvroCoder<GenericRecord> coder;

    RowToAvroBytesFn(Schema beamSchema) {
      avroSchema = toAvroSchema(beamSchema);

View on GitHub (pinned to 12126d8942)

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.

Example fix

// before: blind decode
Row row = bytesToRow.apply(bytes);
// after: guard against empty/malformed input
if (bytes == null || bytes.length == 0) { sendToDeadLetter(bytes); }
else { try { row = bytesToRow.apply(bytes); } catch (AvroRuntimeException e) { log.warn("bad record", e); sendToDeadLetter(bytes); } }

Defensive patterns

Strategy: try-catch

Validate before calling

if (bytes == null || bytes.length == 0) {
  return deadLetter(bytes, "empty payload");
}

Try / catch

try {
  return AvroUtils.getAvroBytesToRowFunction(beamSchema).apply(bytes);
} catch (AvroRuntimeException e) {
  log.warn("Unparseable avro record: {}", e.getMessage());
  return deadLetter(bytes, e); // or use a fallback schema / skip element
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Related errors

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