# Kafka requests

Source: https://apigenie.pl/docs/reference/cli/test/kafka
Part of the reference of [test](https://apigenie.pl/docs/reference/cli/test.md).
Its other pages: [Request files](https://apigenie.pl/docs/reference/cli/test/requests.md), [Variables and OAuth 2](https://apigenie.pl/docs/reference/cli/test/variables.md), [Assertions and scripts](https://apigenie.pl/docs/reference/cli/test/assertions.md), [Validation](https://apigenie.pl/docs/reference/cli/test/validation.md), [Generated requests](https://apigenie.pl/docs/reference/cli/test/generate.md), [GraphQL requests](https://apigenie.pl/docs/reference/cli/test/graphql.md), [WebSocket and MQTT requests](https://apigenie.pl/docs/reference/cli/test/websocket-mqtt.md), [AMQP requests](https://apigenie.pl/docs/reference/cli/test/amqp.md), [gRPC requests](https://apigenie.pl/docs/reference/cli/test/grpc.md), [Importing collections](https://apigenie.pl/docs/reference/cli/test/import.md), [httpYac compatibility](https://apigenie.pl/docs/reference/cli/test/httpyac.md).

A `KAFKA` region produces records to a Kafka cluster and reads what comes on its topics (ADR-170).
The URL names a broker, `kafka://host:port` (port 9092 by default) or `kafkas://host:port` (9093,
over TLS, verified as `https://` with `--ca-cert`, `--insecure` and `--client-cert`); several
bootstrap servers are separated by commas. The cluster needs Kafka 2.3 or later. A cluster that
creates topics on first use creates the region's topic, as Kafka's producers ask it to.

```http
### Place an order: the confirmation carries its correlation id
# @expect 1
# @timeout 10000
# @property correlationId {{$uuid}}
KAFKA kafka://localhost:9092
topic: shop.orders
subscribe: shop.confirmations
key: cust-1

{"sku": "A-1"}

?? body $[0].accepted == true

### Await a shipment the application produces
# @channel shipments
# @expect 1
KAFKA kafka://localhost:9092
subscribe: shop.shipments
```

| Header | Meaning |
| --- | --- |
| `topic` | The topic the records are produced to; also read, unless `subscribe` is given |
| `subscribe` | The topics read, separated by commas |
| `key` | The records' key |
| `partition` | The records' partition; without it, the key's (murmur2, as Kafka's producers choose), else the next in turn |
| `username`, `password` | SASL PLAIN credentials; without them `APIGENIE_KAFKA_USERNAME` and `APIGENIE_KAFKA_PASSWORD`. They are `[redacted]` in the output and reports |

`# @property NAME VALUE` lines are the records' headers, the name as written (`# @property user
NAME VALUE` too). The body's parts (`===`) are the records; `=== wait-for-server` waits for a record
before the next part. Before the first record is produced, the region takes the latest offsets of
the topics it reads, so a reply can't come too early; then it reads until it has `# @expect` records
or `# @timeout` has passed. Its own records are left out. The records read are the response, as an
MQTT region's messages are.

With an AsyncAPI `--spec`, what is produced is checked against the receive operations of the topic's
channel and what is read against its send operations, or as the reply to a request: the value, the
headers against the message's `headers`, the key against the message's Kafka binding `key`
(`message.key.invalid`) and the partition against the channel's Kafka binding `partitions`
(`message.partition.invalid`). A channel's topic is its Kafka binding's `topic`, else its address.
A reply is paired with its request by the correlation id in a header (`$message.header#/...`) or the
payload, and a reply address in a header is followed, as for MQTT 5. Compressed records aren't read;
they are reported as errors.