apache/beam · error · AvroRuntimeException

Could not encode avro from given row: %s

Error message

Could not encode avro from given row: %s

What it means

AvroUtils.getAvroRowToByteFunction returns a function converting a Beam Row to Avro bytes. It first converts the Row via toGenericRecord, then encodes with the Avro coder; any failure (schema mismatch, unsupported field type, encoding error) is wrapped in an AvroRuntimeException with the offending Row rendered via String.format.

Source

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

  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);
      coder = AvroCoder.of(avroSchema);
    }

    @Override
    public byte[] apply(Row row) {
      try {
        GenericRecord record = toGenericRecord(row, avroSchema);
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        coder.encode(record, outputStream);
        return outputStream.toByteArray();
      } catch (Exception e) {
        throw new AvroRuntimeException(
            String.format("Could not encode avro from given row: %s", row), e);
      }
    }
  }

  /**
   * Returns a function mapping AVRO {@link GenericRecord}s to Beam {@link Row}s for use in {@link
   * org.apache.beam.sdk.values.PCollection#setSchema}.
   */
  public static SerializableFunction<GenericRecord, Row> getGenericRecordToRowFunction(
      @Nullable Schema schema) {
    return new GenericRecordToRowFn(schema);
  }

  private static class GenericRecordToRowFn implements SerializableFunction<GenericRecord, Row> {
    private final @Nullable Schema schema;

    GenericRecordToRowFn(@Nullable Schema schema) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Derive the AVRO schema from the current Beam schema (AvroUtils.toAvroSchema(row.getSchema())) instead of passing a hardcoded one.
2. Check for null/non-nullable fields: ensure nullable Beam fields map to AVRO union ["null", T] schemas.
3. Verify all field types have supported conversions (e.g. use logicalType timestamp-micros for datetime fields) and adjust the schema.
4. Update the schema registry subject/version to match the schema actually used for encoding.

Example fix

// before
Schema avroSchema = hardcodedSchema;
byte[] out = rowToBytes.apply(row); // AvroRuntimeException on mismatch
// after
Schema avroSchema = AvroUtils.toAvroSchema(row.getSchema());
byte[] out = AvroUtils.getAvroRowToByteFunction(avroSchema).apply(row);

Schema beamSchema = row.getSchema();
if (avroSchema != null && avroSchema.getFields().size() != beamSchema.getFieldCount()) {
  avroSchema = AvroUtils.toAvroSchema(beamSchema); // avoid guaranteed failure
}

try {
  return avroRowToByteFunction.apply(row);
} catch (AvroRuntimeException e) {
  log.error("Row failed AVRO encoding: {}", row);
  return deadLetter(row, e);
}

When it happens

Trigger: apply(row) invoked on the function returned by AvroUtils.getAvroRowToByteFunction when toGenericRecord(row, avroSchema) fails (e.g. field-count/type mismatch with the supplied schema, unsupported logical types) or coder.encode throws.

Common situations: Beam row schema evolved (fields added/removed/retyped) while a fixed AVRO schema is used for encoding; rows containing logical types (timestamps, decimals) incompatible with the target AVRO schema; writing to sinks (e.g. Kafka with Avro serializer) configured with an outdated schema registry subject.

Related errors

• Cannot provide SerializableCoder because {} does not impleme
• Java Serialization may be non-deterministic.
• cannot encode a null String
• cannot encode a null Integer
• cannot encode a null ValueKind

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