Formats

One produce call per payload type.

producer.produce_many_bytes("events", values=[b"raw", b"bytes"])
producer.produce_many_json("events", values=[{"id": 1}, {"id": 2}])
producer.produce_many_avro("events", values=[{"id": 1}], serde=my_avro_serde)

produce_many_avro encodes the whole batch in Rust against the AvroSerde’s schema.

Schema Registry

For the Confluent wire format ([0x00][schema_id][payload]), use the registry serdes from the core package. They register the schema, look up ids and frame the payload.

from pyrula import AvroRegistrySerde

serde = AvroRegistrySerde.for_topic("http://localhost:8081", "events", SCHEMA)
framed = serde.serialize({"id": 1, "name": "alice"})   # Ok(bytes) with framing
producer.produce_many_bytes("events", values=[framed.unwrap_or_raise()])

unwrap_or_raise() raises the registry error rather than producing an empty record when the registry can’t be reached. JsonSchemaRegistrySerde and ProtobufRegistrySerde cover the other formats; see Serde for details and TLS/mTLS to the registry.

When the registry is down

Every registry serde takes timeout_ms (default 5000, 1 to 300000, else ValueError). It bounds one HTTP request, so a dead registry returns its Err in about that long. The workflows and agents bindings set it with schema_registry_timeout_ms or PYRULA_SCHEMA_REGISTRY_TIMEOUT_MS; the confluent shim takes timeout in seconds on SchemaRegistryClient.

serde = AvroRegistrySerde.for_topic(
    "http://localhost:8081", "events", SCHEMA, timeout_ms=2000
)

A failed lookup is remembered for a fixed 5 seconds: the same schema id or registration gets the same error without a request, then the registry is asked again. register() always asks. This includes a 404 for an unknown id, so a lagging registry follower can fail a brand-new id for up to 5 seconds.

A batch looks up each schema id once. deserialize_many and serialize_many return one Err per record that could not be served; deserialize_many_objects and deserialize_many_from_packed raise on the first failure. If the registry is unreachable or answers 5xx, the batch stops making requests, so an outage costs one timeout_ms. A 4xx belongs to that id alone. Schemas already fetched are still used.

Decoding on consume

Records come back as bytes in record.value. Decode with the matching serde:

for record in batch.records.to_list():
    obj = serde.deserialize(record.value)   # Ok(dict) or Err

Decoding in Rust with poll_arrow

poll_arrow decodes a poll into one Arrow batch with no Python object per record. value_format is "raw" (value bytes in one value column), or "json", "avro" or "avro-confluent" (a column per field of the Avro schema JSON you pass). With include_metadata=True (default) you also get _topic, _partition, _offset, _timestamp and _key.

result = consumer.poll_arrow(value_format="json", schema=SCHEMA)
table = pyarrow.table(result.value.batch)   # result.value.offsets: what to commit

Schema fields must be primitive, date/time, uuid or decimal, optionally nullable. Nested records, arrays, maps and non-null unions raise ConfigError before any record is read.

Not implemented

Nested Avro fields in poll_arrow.

A record that fails to decode loses only itself: the call returns the records before it, the next call returns an error naming its topic, partition and offset and moves past it, and later calls carry on.

"avro" reads each value as one bare datum and needs the exact writer schema; a reader schema with extra fields fails every record. Bytes left after the datum are a decode error, which usually catches a Schema Registry frame (the error points at "avro-confluent"), but a frame that happens to decode as one datum passes. Read registry topics with "avro-confluent".

"avro-confluent" checks the first byte is 0, skips it and the four-byte schema id, and decodes the rest under schema. A wrong first byte, a value under five bytes or trailing bytes are decode errors for that record. The schema id is ignored and the registry is not called, so every value must have been written under schema.

result = consumer.poll_arrow(value_format="avro-confluent", schema=SCHEMA)