Implementation:ArroyoSystems Arroyo Avro Deserializer

Overview

Avro Deserializer provides functions for decoding Apache Avro messages into JSON values, supporting both Confluent Schema Registry wire format and raw Avro datum/container formats. It also handles schema evolution through reader/writer schema resolution.

Description

The module contains two primary functions:

• avro_messages: An async function that decodes Avro binary data into a vector of Avro values. It supports three modes:
  • Confluent Schema Registry: Strips the 5-byte header (magic byte + 4-byte schema ID), resolves the schema from the registry, and decodes the datum.
  • Raw datums: Decodes a single Avro datum using a pre-resolved schema.
  • Embedded schema (container format): Uses the Avro Reader to parse messages with embedded schemas.

• avro_to_json: Recursively converts Avro Value types to serde_json::Value, handling all Avro logical types including timestamps (millis, micros, nanos), dates, durations, UUIDs, decimals, unions, arrays, maps, and records.

The schema registry integration uses a thread-safe HashMap cache protected by a tokio::sync::Mutex to avoid redundant schema resolution.

Usage

This module is called by ArrowDeserializer during message ingestion to convert raw Avro bytes into JSON, which is then converted to Arrow record batches.

Code Reference

Source Location

crates/arroyo-formats/src/avro/de.rs

Signature

pub(crate) async fn avro_messages(
    format: &AvroFormat,
    schema_registry: &Arc<Mutex<HashMap<u32, Schema>>>,
    resolver: &Arc<dyn SchemaResolver + Sync>,
    mut msg: &[u8],
) -> Result<Vec<AvroResult<Value>>, DataflowError>

pub(crate) fn avro_to_json(value: AvroValue) -> JsonValue

Import

use crate::avro::de::{avro_messages, avro_to_json};

I/O Contract

Inputs

Name	Type	Description
format	&AvroFormat	Avro format configuration (confluent_schema_registry, raw_datums, reader_schema)
schema_registry	&Arc<Mutex<HashMap<u32, Schema>>>	Cached schema registry mapping schema IDs to parsed Avro schemas
resolver	&Arc<dyn SchemaResolver + Sync>	Schema resolver for fetching schemas by ID
msg	&[u8]	Raw Avro-encoded message bytes

Outputs

Name	Type	Description
messages	Vec<AvroResult<Value>>	Decoded Avro values (one per record in the message)
json	JsonValue	JSON representation of a single Avro value (from avro_to_json)

Usage Examples

// Decoding a Confluent Schema Registry message
let format = AvroFormat::new(true, false, false);
let schema_registry = Arc::new(Mutex::new(HashMap::new()));
let resolver: Arc<dyn SchemaResolver + Sync> = /* ... */;
let messages = avro_messages(&format, &schema_registry, &resolver, &raw_bytes).await?;

for msg in messages {
    let json_value = avro_to_json(msg?);
    // process json_value...
}

Related Pages

• ArroyoSystems_Arroyo_Avro_Serializer - Complementary Avro serialization
• ArroyoSystems_Arroyo_Avro_Schema_Converter - Schema conversion between Avro and Arrow
• ArroyoSystems_Arroyo_Format_Deserializer - The main ArrowDeserializer that uses this module