# Kafka > Publish events to Kafka topics (plain text or Confluent Avro through a Schema Registry) and assert on the records of a topic. Source: https://axx.nimbusxr.us/references/steps/kafka/ Publish events to Kafka topics (plain text or Confluent Avro through a Schema Registry) and assert on the records of a topic. A **kafka service** names a cluster; a **topic client** gives one topic its producer and consumer configuration, written as Java client properties (`producer.*`, `consumer.*`); **kafka events** are drafts (key, headers, payload) you build and then publish; a **named kafka event** (`the parcel-events kafka event named registered ...`) is an expectation that one record of the topic must meet. Consumer assertions scan every record of the topic from its first offset, re-checking for up to 30 seconds as records arrive. Avro values are compared through Apache Avro's text form of them (GenericData.toString), so JSONPath expectations such as `$.reference` work on Avro records as on JSON; union values appear without their wrapper. **Configuration** (`packs.kafka` in axx.yaml): `timeout` (default `30s`), `maxRecords` kept per topic (default 100000), `lenientUnions` (accept Avro union values without their `{"": value}` wrapper when exactly one branch fits). **Client properties.** Rows prefixed `producer.` or `consumer.` configure that client (without the prefix); values are expanded (`${env:..}`, `${sys:..}`). Defaults: `StringSerializer`/`StringDeserializer`, `auto.offset.reset=earliest`, no consumer group. How each Java property is applied: | Property | Client | In Axx | | --- | --- | --- | | `acks` | producer | `all`/`-1` (default), `1` or `0` (`kgo.RequiredAcks`); `0` and `1` disable idempotence unless it is set explicitly, as in Java. | | `allow.auto.create.topics` | consumer | `true` (default) lets reading a missing topic create it (`kgo.AllowAutoTopicCreation`). Producers always may, as in Java. | | `auto.offset.reset` | consumer | `earliest` (default): assertions consider every record of the topic; `latest`: only records produced after the assertion starts. `none` is an error. | | `bootstrap.servers` | producer, consumer | Seed brokers (`kgo.SeedBrokers`); defaults to the service's `brokers`. | | `buffer.memory` | producer | `kgo.MaxBufferedBytes`. | | `client.id` | producer, consumer | `kgo.ClientID`. | | `compression.type` | producer | `none` (default), `gzip`, `snappy`, `lz4` or `zstd` (`kgo.ProducerBatchCompression`). | | `connections.max.idle.ms` | producer, consumer | `kgo.ConnIdleTimeout`. | | `delivery.timeout.ms` | producer | `kgo.RecordDeliveryTimeout`. | | `enable.idempotence` | producer | `false` sets `kgo.DisableIdempotentWrite`. | | `fetch.max.bytes` | consumer | `kgo.FetchMaxBytes`. | | `fetch.max.wait.ms` | consumer | `kgo.FetchMaxWait`. | | `fetch.min.bytes` | consumer | `kgo.FetchMinBytes`. | | `isolation.level` | consumer | `read_uncommitted` (default) or `read_committed` (`kgo.FetchIsolationLevel`). | | `key.deserializer` | consumer | `StringDeserializer` (default), `ByteArrayDeserializer` or `KafkaAvroDeserializer`. | | `key.serializer` | producer | `StringSerializer` (default), `ByteArraySerializer` (the key text's bytes) or `KafkaAvroSerializer` (the key as an Avro string). | | `linger.ms` | producer | `kgo.ProducerLinger` (default 5 ms, as in Java). | | `max.in.flight.requests.per.connection` | producer | `kgo.MaxProduceRequestsInflightPerBroker`. | | `max.partition.fetch.bytes` | consumer | `kgo.FetchMaxPartitionBytes`. | | `max.request.size` | producer | `kgo.ProducerBatchMaxBytes`. | | `metadata.max.age.ms` | producer, consumer | `kgo.MetadataMaxAge`. | | `partitioner.class` | producer | `DefaultPartitioner` (murmur2 of the key, as by default), `RoundRobinPartitioner` or `UniformStickyPartitioner`; other classes are errors. | | `request.timeout.ms` | producer, consumer | Producer: `kgo.ProduceRequestTimeout`; consumer: `kgo.RequestTimeoutOverhead`. | | `retries` | producer | `kgo.RecordRetries`. | | `retry.backoff.ms` | producer, consumer | Constant `kgo.RetryBackoffFn`. | | `sasl.jaas.config` | producer, consumer | The `username` and `password` of a `PlainLoginModule` or `ScramLoginModule` entry. | | `sasl.mechanism` | producer, consumer | `PLAIN`, `SCRAM-SHA-256` or `SCRAM-SHA-512` (GSSAPI and OAUTHBEARER are not supported). | | `security.protocol` | producer, consumer | `PLAINTEXT`, `SSL` (TLS dialer), `SASL_PLAINTEXT` or `SASL_SSL` (`kgo.SASL`). | | `socket.connection.setup.timeout.ms` | producer, consumer | `kgo.DialTimeout`. | | `ssl.enabled.protocols` | producer, consumer | Limits TLS versions to the listed `TLSv1.2`/`TLSv1.3`. | | `ssl.endpoint.identification.algorithm` | producer, consumer | `https` (default) verifies the broker host name; empty skips that check (the chain is still verified). | | `ssl.key.password` | producer, consumer | Private key password (JKS key entries, encrypted PEM keys); defaults to the keystore password. | | `ssl.keystore.certificate.chain` | producer, consumer | Inline PEM certificate chain. | | `ssl.keystore.key` | producer, consumer | Inline PEM private key (with `ssl.keystore.certificate.chain`). | | `ssl.keystore.location` | producer, consumer | Client certificate and key (JKS, PKCS12 or PEM file) for mutual TLS. | | `ssl.keystore.password` | producer, consumer | Keystore password. | | `ssl.keystore.type` | producer, consumer | `JKS` (default), `PKCS12` or `PEM`. | | `ssl.protocol` | producer, consumer | `TLSv1.2` or `TLSv1.3` sets the minimum TLS version (`TLS` allows both). | | `ssl.truststore.certificates` | producer, consumer | Inline PEM CA certificates. | | `ssl.truststore.location` | producer, consumer | CA certificates (JKS, PKCS12 or PEM file, resolved against `resources`). | | `ssl.truststore.password` | producer, consumer | Truststore password (optional for JKS, as in Java). | | `ssl.truststore.type` | producer, consumer | `JKS` (default), `PKCS12` or `PEM`; JKS and PKCS12 files are recognized by content. | | `transactional.id` | producer | Error: the steps never begin a transaction, so a transactional producer cannot send. | | `value.deserializer` | consumer | `StringDeserializer` (default), `ByteArrayDeserializer` or `KafkaAvroDeserializer` (payloads are checked against the record's Java `toString()`). | | `value.serializer` | producer | `StringSerializer` (default), `ByteArraySerializer` or `KafkaAvroSerializer` (needed to publish with a schema). | | `auto.register.schemas` | producer | `true` (default) registers the schema under the subject; `false` looks its ID up and fails if it is not registered. | | `basic.auth.credentials.source` | producer, consumer | `URL` (default), `USER_INFO` or `SASL_INHERIT` (the SASL username and password). | | `basic.auth.user.info` | producer, consumer | `user:password` for `USER_INFO`. | | `bearer.auth.credentials.source` | producer, consumer | Only `STATIC_TOKEN` is supported. | | `bearer.auth.token` | producer, consumer | Static bearer token for the registry. | | `key.subject.name.strategy` | producer | `TopicNameStrategy` (default: `-key`), `RecordNameStrategy` or `TopicRecordNameStrategy`. | | `normalize.schemas` | producer | Passes `normalize=true` when registering or looking up. | | `schema.reflection` | producer, consumer | Only `false`: reflection needs Java classes. | | `schema.registry.basic.auth.user.info` | producer, consumer | Older name of `basic.auth.user.info`. | | `schema.registry.url` | producer, consumer | Schema Registry URLs (comma-separated); required by the Avro (de)serializers. `user:password@` in a URL is used for basic auth. | | `specific.avro.reader` | consumer | Only `false`: Axx has no generated classes and reads generic records. | | `use.latest.version` | producer | With `auto.register.schemas=false`, writes with the subject's latest schema and ID. | | `use.schema.id` | producer | Writes with this schema ID (with `auto.register.schemas=false`). | | `value.subject.name.strategy` | producer | Same choices; the default subject is `-value`. | | `schema.registry.ssl.*` | producer, consumer | The `ssl.*` settings above, for HTTPS to the Schema Registry. | Accepted without effect: `group.id`, `group.instance.id`, `group.protocol`, `group.remote.assignor`, `enable.auto.commit`, `auto.commit.interval.ms`, `session.timeout.ms`, `heartbeat.interval.ms`, `max.poll.interval.ms`, `max.poll.records`, `partition.assignment.strategy`, `internal.leave.group.on.close`, `exclude.internal.topics`, `default.api.timeout.ms`, `client.rack`, `check.crcs`, `internal.throw.on.fetch.stable.offset.unsupported` (Axx reads each topic from the start without a consumer group and never commits offsets). `batch.size`, `max.block.ms`, `metadata.max.idle.ms`, `partitioner.ignore.keys`, `partitioner.adaptive.partitioning.enable`, `partitioner.availability.timeout.ms`, `transaction.timeout.ms`, `compression.gzip.level`, `compression.lz4.level`, `compression.zstd.level` (franz-go sizes batches by `max.request.size` and publishes each event synchronously). `client.dns.lookup`, `receive.buffer.bytes`, `send.buffer.bytes`, `reconnect.backoff.ms`, `reconnect.backoff.max.ms`, `retry.backoff.max.ms`, `socket.connection.setup.timeout.max.ms`, `metadata.recovery.strategy`, `metrics.num.samples`, `metrics.recording.level`, `metrics.sample.window.ms`, `auto.include.jmx.reporter`, `enable.metrics.push`, `ssl.provider`, `ssl.cipher.suites`, `ssl.keymanager.algorithm`, `ssl.trustmanager.algorithm`, `ssl.secure.random.implementation`, `sasl.kerberos.service.name`, `sasl.login.connect.timeout.ms`, `sasl.login.read.timeout.ms`, `sasl.login.retry.backoff.ms`, `sasl.login.retry.backoff.max.ms`, `sasl.login.refresh.window.factor`, `sasl.login.refresh.window.jitter`, `sasl.login.refresh.min.period.seconds`, `sasl.login.refresh.buffer.seconds`, `key.serializer.encoding`, `value.serializer.encoding`, `serializer.encoding`, `key.deserializer.encoding`, `value.deserializer.encoding`, `deserializer.encoding` (Java tuning without a franz-go counterpart (strings are always UTF-8)). `latest.compatibility.strict`, `id.compatibility.strict`, `avro.remove.java.properties`, `avro.use.logical.type.converters`, `avro.reflection.allow.null`, `max.schemas.per.subject`, `use.latest.with.metadata`, `auto.register.schemas.retry` (Axx writes and reads generic Avro records as described above). Rejected (they name Java classes): `context.name.strategy`, `interceptor.classes`, `sasl.client.callback.handler.class`, `sasl.login.callback.handler.class`, `sasl.login.class`, `security.providers`, `specific.avro.key.type`, `specific.avro.value.type`, `ssl.engine.factory.class`, `metric.reporters` other than JmxReporter, and any unknown `*.class`/`*.classes` property. Other unknown properties are logged as warnings and ignored. ## `kafka.service` ```gherkin Given the {word} kafka service with the following properties: | ... | ... | ``` Register a Kafka cluster. The first one registered in a scenario is the default for steps without `on the {word} kafka service`. Properties: `brokers` (required; `host:port` list, `${env:..}`/`${sys:..}` expanded). **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Given the events kafka service with the following properties: ``` ## `kafka.client` ```gherkin Given the {word} kafka topic client ``` Create a topic client on the default Kafka service with the default configuration: string keys and values, records read from the start of the topic. A topic can have one client per service in a scenario. **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Given the parcel-events kafka topic client ``` ## `kafka.client.props` ```gherkin Given a(n) {word} kafka topic client[[ on the {word} kafka service]] with the following properties: | ... | ... | ``` Create a topic client configured with Java Kafka client properties. Rows prefixed `producer.` configure publishing, rows prefixed `consumer.` configure assertions (the prefix is removed); values are expanded. For Avro use `producer.value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer`, `consumer.value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer` and `*.schema.registry.url`. See the pack documentation for every supported property; unknown properties are logged as warnings. **Variants** (optional parts in `[[...]]` above): - `a(n) {word} kafka topic client with the following properties:` - `a(n) {word} kafka topic client on the {word} kafka service with the following properties:` **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Given a depot-scans kafka topic client with the following properties: Given a parcel-events kafka topic client on the events kafka service with the following properties: ``` ## `kafka.event` ```gherkin Given a(n)[[ {ordinal} ordered]] {word} kafka event[[ on {word} kafka service]] ``` Draft a new event (key, headers and payload are set by the following steps; the payload starts as `{}`). Without an ordinal the event is appended. With one, it must be the next position (`a 2nd ordered` after one event); an ordinal equal to the number of existing events still appends, with a warning (it will be an error in Axx 1.0). **Variants** (optional parts in `[[...]]` above): - `a(n) {word} kafka event` - `a(n) {ordinal} ordered {word} kafka event` - `a(n) {word} kafka event on {word} kafka service` - `a(n) {ordinal} ordered {word} kafka event on {word} kafka service` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces) **Example:** ```gherkin Given a depot-scans kafka event Given a 2nd ordered depot-scans kafka event Given a depot-scans kafka event on events kafka service ``` ## `kafka.event.key` ```gherkin Given the[[ {ordinal} ordered]] {word} kafka event key is {word}[[ on the {word} kafka service]] ``` Set the key of a drafted event. With `the {ordinal} ordered` the step works on that event of the topic (`1st` is the first event created); without it, on the first event. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event key is {word}` - `the {ordinal} ordered {word} kafka event key is {word}` - `the {word} kafka event key is {word} on the {word} kafka service` - `the {ordinal} ordered {word} kafka event key is {word} on the {word} kafka service` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces) **Example:** ```gherkin Given the depot-scans kafka event key is PX-1001 Given the 2nd ordered depot-scans kafka event key is PX-1002 on the events kafka service ``` ## `kafka.event.headers` ```gherkin Given the[[ {ordinal} ordered]] {word} kafka event headers[[ on the {word} kafka service]] are: | ... | ... | ``` Set headers of a drafted event (`name | value` rows; setting a header again replaces its value; an empty cell sends the text `null`). With `the {ordinal} ordered` the step works on that event of the topic (`1st` is the first event created); without it, on the first event. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event headers are:` - `the {ordinal} ordered {word} kafka event headers are:` - `the {word} kafka event headers on the {word} kafka service are:` - `the {ordinal} ordered {word} kafka event headers on the {word} kafka service are:` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces) **Example:** ```gherkin Given the depot-scans kafka event headers are: Given the 1st ordered depot-scans kafka event headers on the events kafka service are: ``` ## `kafka.event.payload.resource` ```gherkin Given the[[ {ordinal} ordered]] {word} kafka event payload is a(n) {filepath} resource[[ on the {word} kafka service]] ``` Set the payload of a drafted event to the contents of a file (resolved against `resources`). For Avro events the file is Avro's JSON encoding of the record (unions as `{"": value}`). With `the {ordinal} ordered` the step works on that event of the topic (`1st` is the first event created); without it, on the first event. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event payload is a(n) {filepath} resource` - `the {ordinal} ordered {word} kafka event payload is a(n) {filepath} resource` - `the {word} kafka event payload is a(n) {filepath} resource on the {word} kafka service` - `the {ordinal} ordered {word} kafka event payload is a(n) {filepath} resource on the {word} kafka service` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces), `{filepath}` (A file of the project, without whitespace: a path relative to the `resources` directories or to the directory of axx.yaml, or an absolute path. Editors link it to the file) **Example:** ```gherkin Given the depot-scans kafka event payload is a kafka/scan-delivered.json resource Given the 3rd ordered depot-scans kafka event payload is a kafka/scan-out-for-delivery.json resource on the events kafka service ``` ## `kafka.event.properties.first` ```gherkin Given the[[ {ordinal} ordered]] kafka event payload properties[[ on the {word} kafka service]] are: | ... | ... | ``` Set JSONPath properties (`path | value` rows) of an event of the service's **first** topic client (the first one created in the scenario). Values are always set as strings, an empty cell sets JSON null, and every property must already exist in the payload. With `the {ordinal} ordered` the step works on that event of the topic (`1st` is the first event created); without it, on the first event. **Variants** (optional parts in `[[...]]` above): - `the kafka event payload properties are:` - `the {ordinal} ordered kafka event payload properties are:` - `the kafka event payload properties on the {word} kafka service are:` - `the {ordinal} ordered kafka event payload properties on the {word} kafka service are:` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces) **Example:** ```gherkin Given the kafka event payload properties are: Given the 2nd ordered kafka event payload properties on the events kafka service are: ``` ## `kafka.event.properties` ```gherkin Given the {word} kafka event payload properties[[ on the {word} kafka service]] are: | ... | ... | ``` Set JSONPath properties (`path | value` rows) of the topic's first event. Values are always set as strings, an empty cell sets JSON null, and every property must already exist in the payload (set it in the payload file first). **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event payload properties are:` - `the {word} kafka event payload properties on the {word} kafka service are:` **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Given the depot-scans kafka event payload properties are: Given the depot-scans kafka event payload properties on the events kafka service are: ``` ## `kafka.event.properties.ordinal` ```gherkin Given the {ordinal} ordered {word} kafka event payload properties[[ on the {word} kafka service]] are: | ... | ... | ``` Like the topic form, for the given event of the topic (`1st` is the first event created). **Variants** (optional parts in `[[...]]` above): - `the {ordinal} ordered {word} kafka event payload properties are:` - `the {ordinal} ordered {word} kafka event payload properties on the {word} kafka service are:` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces) **Example:** ```gherkin Given the 2nd ordered depot-scans kafka event payload properties are: ``` _Since 0.1.0._ ## `kafka.event.property.null` ```gherkin Given the[[ {ordinal} ordered]] {word} kafka event payload property {word} is null[[ on the {word} kafka service]] ``` Set a JSONPath property of a drafted event's payload to JSON null; the property must exist. With `the {ordinal} ordered` the step works on that event of the topic (`1st` is the first event created); without it, on the first event. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event payload property {word} is null` - `the {ordinal} ordered {word} kafka event payload property {word} is null` - `the {word} kafka event payload property {word} is null on the {word} kafka service` - `the {ordinal} ordered {word} kafka event payload property {word} is null on the {word} kafka service` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces) **Example:** ```gherkin Given the depot-scans kafka event payload property $.location is null Given the 2nd ordered depot-scans kafka event payload property $.location is null on the events kafka service ``` ## `kafka.event.publish.schema` ```gherkin When the[[ {ordinal} ordered]] {word} kafka event is published using schema {filepath}[[ on the {word} kafka service]] ``` Publish a drafted event as Confluent Avro: the payload (Avro's JSON encoding) is read with the `.avsc` schema file, the schema is registered (or looked up) in the Schema Registry under the subject of `value.subject.name.strategy` (`-value` by default), and the record is written as magic byte 0, the schema ID and the Avro binary. Needs `producer.value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer` and `producer.schema.registry.url`. A payload that does not fit the schema fails with the JSONPath of the mismatch. With `the {ordinal} ordered` the step works on that event of the topic (`1st` is the first event created); without it, on the first event. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event is published using schema {filepath}` - `the {ordinal} ordered {word} kafka event is published using schema {filepath}` - `the {word} kafka event is published using schema {filepath} on the {word} kafka service` - `the {ordinal} ordered {word} kafka event is published using schema {filepath} on the {word} kafka service` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces), `{filepath}` (A file of the project, without whitespace: a path relative to the `resources` directories or to the directory of axx.yaml, or an absolute path. Editors link it to the file) **Example:** ```gherkin When the depot-scans kafka event is published using schema schemas/depot-scan.avsc When the 2nd ordered depot-scans kafka event is published using schema schemas/depot-scan.avsc on the events kafka service ``` ## `kafka.event.publish` ```gherkin When the[[ {ordinal} ordered]] {word} kafka event is published[[ on the {word} kafka service]] ``` Publish a drafted event as it is: the payload text with the producer's value serializer (`StringSerializer` by default, or `ByteArraySerializer`), with its key and headers. Use `published using schema` for Avro. With `the {ordinal} ordered` the step works on that event of the topic (`1st` is the first event created); without it, on the first event. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event is published` - `the {ordinal} ordered {word} kafka event is published` - `the {word} kafka event is published on the {word} kafka service` - `the {ordinal} ordered {word} kafka event is published on the {word} kafka service` **Parameters:** `{ordinal}` (A 1-based position such as `1st`, `2nd`, `3rd` or `4th`. Omitting an optional ordinal means the first), `{word}` (one word, no spaces) **Example:** ```gherkin When the depot-scans kafka event is published When the 2nd ordered depot-scans kafka event is published on the events kafka service ``` _Since 0.1.0._ ## `kafka.consumed.key` ```gherkin Then the {word} kafka event named {word} key is {word}[[ on the {word} kafka service]] ``` Expect the label's record to have this key (the consumer's key deserializer decides how keys read). Adds the expectation to the label (`named {word}`) and then waits until **one record** of the topic satisfies **every** expectation added to that label in the scenario (key, payload properties and headers together). Records are read from the start of the topic (or, with `consumer.auto.offset.reset=latest`, only those produced after the step starts); the step fails after 30 seconds (`packs.kafka.timeout`) without a match, and the failure report lists the label's expectations and the latest records with the reason each one did not match. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event named {word} key is {word}` - `the {word} kafka event named {word} key is {word} on the {word} kafka service` **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Then the parcel-events kafka event named registered key is PX-1001 Then the parcel-events kafka event named registered key is PX-1001 on the events kafka service ``` ## `kafka.consumed.properties` ```gherkin Then the {word} kafka event named {word} payload properties[[ on the {word} kafka service]] are: | ... | ... | ``` Expect JSONPath properties of the label's record payload (`path | value` rows). Values are typed: `"text"` is a string, `null` is JSON null, `12` an integer, `1.5` a decimal, `true`/`false` booleans, `{...}`/`[...]` JSON; anything else is a string. Numbers must match in type (`2` does not equal `2.0`). Adds the expectation to the label (`named {word}`) and then waits until **one record** of the topic satisfies **every** expectation added to that label in the scenario (key, payload properties and headers together). Records are read from the start of the topic (or, with `consumer.auto.offset.reset=latest`, only those produced after the step starts); the step fails after 30 seconds (`packs.kafka.timeout`) without a match, and the failure report lists the label's expectations and the latest records with the reason each one did not match. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event named {word} payload properties are:` - `the {word} kafka event named {word} payload properties on the {word} kafka service are:` **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Then the parcel-events kafka event named registered payload properties are: Then the parcel-events kafka event named registered payload properties on the events kafka service are: ``` ## `kafka.consumed.headers` ```gherkin Then the {word} kafka event named {word} headers[[ on the {word} kafka service]] are: | ... | ... | ``` Expect headers of the label's record (`name | value` rows): each header must occur exactly once with exactly this value. Adds the expectation to the label (`named {word}`) and then waits until **one record** of the topic satisfies **every** expectation added to that label in the scenario (key, payload properties and headers together). Records are read from the start of the topic (or, with `consumer.auto.offset.reset=latest`, only those produced after the step starts); the step fails after 30 seconds (`packs.kafka.timeout`) without a match, and the failure report lists the label's expectations and the latest records with the reason each one did not match. **Variants** (optional parts in `[[...]]` above): - `the {word} kafka event named {word} headers are:` - `the {word} kafka event named {word} headers on the {word} kafka service are:` **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Then the parcel-events kafka event named registered headers are: Then the parcel-events kafka event named registered headers on the events kafka service are: ``` ## `kafka.consumed.headers.match` ```gherkin Then the {word} kafka event named {word} headers match: | ... | ... | ``` Expect headers of the label's record to match regular expressions (`name | pattern` rows, Java syntax, whole value). A header must have one distinct value; this step (unlike the others) ignores repeated identical values of a header. Adds the expectation to the label (`named {word}`) and then waits until **one record** of the topic satisfies **every** expectation added to that label in the scenario (key, payload properties and headers together). Records are read from the start of the topic (or, with `consumer.auto.offset.reset=latest`, only those produced after the step starts); the step fails after 30 seconds (`packs.kafka.timeout`) without a match, and the failure report lists the label's expectations and the latest records with the reason each one did not match. **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Then the parcel-events kafka event named registered headers match: ``` ## `kafka.consumed.headers.match.service` ```gherkin Then the {word} kafka event named {word} headers on the {word} kafka service match: | ... | ... | ``` The `headers match` expectation for a topic client of a named Kafka service. Adds the expectation to the label (`named {word}`) and then waits until **one record** of the topic satisfies **every** expectation added to that label in the scenario (key, payload properties and headers together). Records are read from the start of the topic (or, with `consumer.auto.offset.reset=latest`, only those produced after the step starts); the step fails after 30 seconds (`packs.kafka.timeout`) without a match, and the failure report lists the label's expectations and the latest records with the reason each one did not match. **Parameters:** `{word}` (one word, no spaces) **Example:** ```gherkin Then the parcel-events kafka event named registered headers on the events kafka service match: ``` _Since 0.1.0._