Example: Kafka

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

Kafka: publish, stream, deliver

The whole path of Publishing and consuming on a broker on Kafka: a publication door, also an MCP tool; an SSE fan-out or acknowledged WebSocket stream, bound to the tenant; acme's events delivered as signed webhooks, dead letters in orders.dead. The gateway authenticates with SASL SCRAM over TLS.

# Kafka behind the gateway's controls (1.5.0): an HTTP door and an MCP tool publish, a
# browser streams its tenant's events in SSE or WebSocket, a partner receives them as signed webhooks.
# flexway.yaml lists no channel - they come from the AsyncAPI contract.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml                      # AsyncAPI 3.0.0 or 3.1.0
upstreams:
  events:
    protocol: kafka
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093]
    tls: { server_name: kafka.internal }          # TLS, system roots; ca_file, cert_file and key_file for a private CA or mTLS
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
    timeout: 2s                                  # one publication's budget, the acknowledgement included
    max_in_flight: 512                           # bounded buffer: past it, 503 overloaded
  partner-hook:
    protocol: webhook
    url: https://hooks.partner.example/orders
    secrets: ["file:/etc/flexway-secrets/hooks/current", "file:/etc/flexway-secrets/hooks/next"]
    retries: { max: 8, max_delay: 10m }
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org                          # the tenant every channel below is bound to
agents:
  mcp:
    resources:
      api.example.com: https://api.example.com/mcp

# examples/async-kafka/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders events, version: 1.0.0 }
channels:
  orderCreated:
    address: "orders.{tenant}.created"           # one topic per tenant
    parameters:
      tenant: { x-flexway-bind: tenant }         # bound to the verified identity, never read from the request
    messages:
      created:
        contentType: application/json
        payload:
          type: object
          required: [orderId]
          properties:
            orderId: { type: string, minLength: 1 }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-key: /orderId                      # ordered per key, and per key only
  receiveOrderCreated:
    action: receive
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-deliver:                           # acme's events to its webhook; what fails ends in orders.dead
      - { upstream: partner-hook, tenant: acme, dead_letter: orders.dead }

# examples/async-kafka/openapi.yaml
openapi: 3.1.0
info: { title: Orders events, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      summary: Publish an order-created event for the caller's tenant
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }
      x-flexway-tool: { name: emit_order_created }   # the same door, offered to agents as an MCP tool
      security: [ { customers: [orders:write] } ]
      requestBody:
        required: true
        content:
          application/json:
            schema: { type: object, required: [orderId], properties: { orderId: { type: string, minLength: 1 } } }
      responses: { "202": { description: Acknowledged by the broker } }
  /orders/stream:
    get:
      operationId: streamOrderCreated
      x-flexway-subscribe: { operation: receiveOrderCreated, transport: [sse, websocket], max_streams: 2, ack_timeout: 30s, buffer: 256, history: 64 }
      security: [ { customers: [orders:read] } ]
      responses: { "200": { description: The caller's tenant's events } }

Case	What the gateway does
POST /orders/events or the emit_order_created tool	202 after acks=all; Idempotency-Key required (428 otherwise), becoming the id; ordered by orderId
GET /orders/stream in SSE	One consumer per channel and replica; Last-Event-ID resumed from the last 64 messages, 412 precondition_failed beyond
GET /orders/stream in WebSocket	A group of the subscriber's own, {"ack": "<id>"} within 30 s, otherwise a 1008 close and a new delivery
Delivery to partner-hook	A group shared by the replicas, Standard Webhooks, position committed after the 2xx, dead letters in orders.dead
mTLS instead of SASL	tls.cert_file and tls.key_file mounted, no sasl

Run

1. Save the flexway.yaml shown and the contracts it references in one directory, next to it (examples/async-kafka/ 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/...), then start:

   docker run --rm -p 8080:8080 -p 9090:9090 \
     -v "$PWD:/etc/flexway:ro" -v "$PWD/secrets:/etc/flexway-secrets:ro" \
     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.