# Example: Kafka > Kafka behind the gateway: publish from HTTP or MCP, stream in SSE or WebSocket, deliver as signed webhooks. Example - Since version 1.5.0 - Flexway v1.7.1 - https://flexway.io/en/docs/operations-examples-kafka/ > 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/). ## Kafka: publish, stream, deliver The whole path of [Publishing and consuming on a broker](https://flexway.io/en/docs/operations-async/) 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. ```yaml # 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 ``` ```yaml # 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 } ``` ```yaml # 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": ""}` 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: ```bash docker run --rm -v "$PWD:/etc/flexway:ro" ghcr.io/flexwiz/flexway: validate -v ``` 3. Mount the secrets the file references (`file:/etc/flexway-secrets/...`), then start: ```bash 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: ``` ## 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":}`; 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/).