Example: Avro payloads and schema registry

Operations guide: configure, deploy and observe the gateway. What the feature does, product side: product guide; calling it: developer guide.

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.

# 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

# 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

# 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

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:

   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:

   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, Receive signed webhooks.