# Example: Avro payloads and schema registry

> Avro orders checked under their schema, read as JSON or as they are, the schema registry's wire format.

Example - Since version 1.6.0 - Flexway v1.7.1 - https://flexway.io/en/docs/operations-examples-avro/

> Operations guide: configure, deploy and observe the gateway. What the feature does, product side: [product guide](https://flexway.io/en/docs/product-overview/); calling it: [developer guide](https://flexway.io/en/docs/developer-overview/).

## Avro payloads under schema, schema registry

A shop publishes its orders in **Avro** from HTTP: the body is decoded under `order.avsc`, compiled at
load, before it reaches Kafka; a dashboard reads them over SSE **in JSON**, a service reads them over a
WebSocket **in Avro, as they are**. The audit channel carries **Confluent's wire format**: each schema
id is read from the registry, and must be registered there under the tenant's subject
(`<tenant>.orders.audit-value`) - another tenant never publishes under that schema.

<!-- example: examples/async-binary/flexway.yaml -->

```yaml
# Avro orders through the asynchronous plane (1.6.0): an HTTP caller publishes an Avro
# body - checked under its schema, compiled at load -, a subscriber over SSE or a webhook reads it in
# JSON, a WebSocket that declares binary frames reads it as it is. The audit channel carries Confluent's
# wire format: each schema id is read in the registry and must be registered under the tenant's subject.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-brokers:
    protocol: kafka
    brokers: [kafka-0.internal:9092, kafka-1.internal:9092]
schema_registry:
  url: https://registry.internal:8081                # read, never written - the Confluent API
  auth:
    username: flexway
    password: ${REGISTRY_PASSWORD}                    # a reference, resolved at each request
  timeout: 2s                                         # a lookup holds the message that needs it
  max_schemas: 1000                                   # compiled once each, kept by id
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org
```

```yaml
# examples/async-binary/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Shop orders, version: "1.0.0" }
channels:
  orders:
    address: "{tenant}.orders"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      order:
        contentType: application/vnd.apache.avro+binary
        payload: { schemaFormat: "application/vnd.apache.avro;version=1.9.0", schema: { $ref: "./order.avsc" } }
  audit:
    address: "{tenant}.orders.audit"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      audited:
        contentType: application/vnd.apache.avro+binary
        payload: { schemaFormat: "application/vnd.apache.avro;version=1.9.0", schema: { $ref: "./order.avsc" } }
        bindings:
          kafka: { schemaIdLocation: payload, schemaIdPayloadEncoding: confluent, schemaLookupStrategy: TopicIdStrategy }
operations:
  publishOrder:
    action: send
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: orders-brokers
    x-flexway-validation: full          # decode every body under order.avsc - opt-in for a binary message
    x-flexway-key: /orderId             # read in the payload's JSON form
  receiveOrder:
    action: receive
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: orders-brokers
  publishAudit:
    action: send
    channel: { $ref: "#/channels/audit" }
    x-flexway-upstream: orders-brokers
    x-flexway-validation: full          # the schema id in the registry, under <tenant>.orders.audit-value
  receiveAudit:
    # Records of an ordinary Kafka producer (a Confluent serializer, no CloudEvents header) are read
    # too: the channel is bound to the tenant and its only message is in the registry's wire format,
    # so the gateway builds each envelope - id <topic>/<partition>/<offset>, the record's timestamp.
    action: receive
    channel: { $ref: "#/channels/audit" }
    x-flexway-upstream: orders-brokers
```

```yaml
# examples/async-binary/openapi.yaml
openapi: 3.1.0
info: { title: Shop orders, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
security: [{ customers: [] }]
paths:
  /orders/events:
    post:
      operationId: emitOrder
      x-flexway-publish: { operation: publishOrder, idempotency: required }
      requestBody:
        required: true
        content: { application/vnd.apache.avro+binary: { schema: { type: string, format: binary } } }
      responses: { "202": { description: published, after the broker's acknowledgement } }
  /orders/audit:
    post:
      operationId: emitAudit
      x-flexway-publish: { operation: publishAudit }
      requestBody:
        required: true
        content: { application/vnd.apache.avro+binary: { schema: { type: string, format: binary } } }
      responses: { "202": { description: published } }
  /orders/stream:
    get:
      operationId: streamOrders
      x-flexway-subscribe: { operation: receiveOrder, transport: [sse] }
      responses: { "200": { description: the tenant's orders, in JSON } }
  /orders/frames:
    get:
      operationId: framesOrders
      x-flexway-subscribe: { operation: receiveOrder, transport: [websocket], frames: binary }
      responses: { "200": { description: the tenant's orders, the Avro payload in binary frames } }
  /orders/audit/stream:
    get:
      operationId: streamAudit
      x-flexway-subscribe: { operation: receiveAudit, transport: [sse] }
      responses: { "200": { description: the tenant's audit records in JSON, ordinary producers' included } }
```

| Case | What the gateway does |
| --- | --- |
| `POST /orders/events`, a conforming Avro body | `202` after the acknowledgement; the Kafka key read at `/orderId` of the message's JSON form |
| A body that does not decode under `order.avsc` | `422 invalid_request`, the rule alone, never a value of the message |
| Another `Content-Type` | `422 invalid_request` (the door declares Avro only); a door without `requestBody` answers `415 unsupported_media_type` |
| `POST /orders/audit`, an id registered under `acme.orders.audit-value` | `202`; the schema read once per id, kept in memory (`max_schemas`) |
| An unknown id, or one registered under another subject | `422 invalid_request` |
| The registry unreachable or failing | `503 upstream_unavailable` with `Retry-After`: never a message published unvalidated; a breaker of the registry's own |
| `GET /orders/stream` (SSE) | Each event in JSON (Avro's JSON encoding), `datacontenttype: application/json`; a message that does not decode is not handed on (`flexway_async_deliver_total{outcome="invalid"}`) |
| `GET /orders/frames` (WebSocket) | A text frame (the attributes, the `id` to acknowledge among them), then a **binary** frame: the Avro payload as it is |
| Webhook (`x-flexway-deliver`) | A CloudEvent body with the data transcoded to JSON; the registry unreachable: the message is neither delivered nor acknowledged, it comes again |
| Protobuf | `schemaFormat: "application/vnd.google.protobuf;version=3"` and `schema: { $ref: "./reading.binpb#acme.Reading" }` - a compiled descriptor set (`protoc --descriptor_set_out --include_imports`, `buf build`), the message by its name; inline `.proto` sources are refused |
| CBOR | `contentType: application/cbor` and a JSON schema: the payload is read in CBOR's JSON model (RFC 8949 section 6.1) and checked against that schema |
| `GET /orders/audit/stream`, a record of an **ordinary Kafka producer** (wire format, no CloudEvents header) | Handed on in the envelope the gateway builds: `id` = `acme.orders.audit/<partition>/<offset>` (the same on every replica: the SSE `id` and `Last-Event-ID`, the `id` a WebSocket acknowledges, the `webhook-id`), `source` = `/flexway/orders-brokers/audit`, `type` = `audited`, `time` = the Kafka timestamp, `datacontenttype: application/json` after transcoding; **no** `tenant` attribute: the tenant is the bound channel's, never a header nor the message's content |
| A record without attributes on another channel, or one that carries a `ce_*` attribute without being an event | Not an event, as in `1.5.0`: not handed on (`flexway_async_deliver_total{outcome="invalid"}`) |
| The registry's writer schema differs from the contract's | **Resolved** to the contract's schema by Avro's standard rules (a field added with a default, a field removed, `int` -> `long`, aliases, an enum's default symbol): the same JSON as a message written under the contract's schema; resolved once per pair (schema id, contract schema), `max_schemas` pairs kept. Identical schemas (one canonical form) cost nothing more |
| The writer schema does not resolve to the contract's (a field without a default, an incompatible type) | Refused whole, never handed on in part: `422 invalid_request` on publication; on consumption not handed on, as a payload that does not decode - SSE at most once, WebSocket acknowledged, webhook dead-lettered -, the consumption goes on; counted `outcome="incompatible"` |
| `x-flexway-key` on a binary message | The key is read in the message's **JSON** form: **one decoding per publication**, even with `x-flexway-validation: off` (a payload that does not decode is then refused with `422`) - a cost to weigh at a high rate; without a key, a payload not validated is relayed unread |

> [!WARNING]
> Refused at load: an unknown `schemaFormat` (RAML, OpenAPI's, Avro `2.x`), an Avro schema that is
> invalid, recursive, or with an array whose items take no byte, a message absent from the descriptor
> set, a binary schema `$ref` outside the AsyncAPI document's directory, the wire format without
> `schema_registry` or on a broker that is not Kafka, `x-flexway-validation: off` on a JSON message (always
> checked), `frames: binary` without a WebSocket or on a JSON message, a consumption of a wire-format
> channel isolated by the `tenant` attribute alone (neither bound nor public): an ordinary producer writes
> no attribute, and a tenant is never read from a message.

### Measure

 `flexway_async_payloads_total{operation, format, step, outcome}` - `step` is `validate` or
`transcode`, `outcome` `ok`, `invalid`, `not_attached`, `incompatible` or `unavailable`; never a value of the message
nor a schema id.

## Run

1. Save the `flexway.yaml` shown and the contracts it references in one directory, next to it (`examples/async-binary/` in the release sources).
2. Validate offline, without a broker or a network - the validator ships in the published image:

   ```bash
   docker run --rm -v "$PWD:/etc/flexway:ro" ghcr.io/flexwiz/flexway:<version> validate -v
   ```

3. Mount the secrets the file references (`file:/etc/flexway-secrets/...`, environment variables), then start:

   ```bash
   docker run --rm -p 8080:8080 -p 9090:9090 \
     -v "$PWD:/etc/flexway:ro" -v "$PWD/secrets:/etc/flexway-secrets:ro" -e REGISTRY_PASSWORD \
     ghcr.io/flexwiz/flexway:<version>
   ```

## Verify

1. `validate -v` exits `0` and lists every resolved operation; a refusal names the faulty field and exits `1`.
2. Once started, `curl -s localhost:9090/readyz` answers `200` `{"status":"ready","operations":<n>}`; a missing secret makes the start refuse, naming it.
3. Each row of the case table replays from the developer side: [Publish and subscribe](https://flexway.io/en/docs/developer-async/), [Receive signed webhooks](https://flexway.io/en/docs/developer-webhooks/).