# Formats

> Produce bytes, JSON, or Avro, and frame for Confluent Schema Registry.

Source: https://docs.pyrula.io/kafka/formats/

One produce call per payload type.

```python
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.

```python
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](https://docs.pyrula.io/primitives/serde.md) 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`.

```python
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:

```python
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`.

```python
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.

:::note[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`.

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