Kafka requests

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.

### 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.