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 {"<branch>": value} wrapper when exactly one branch fits).

Client properties. A topic client’s table takes Java Kafka client properties, prefixed producer. or consumer.: a(n) {word} kafka topic client with the following properties: lists those Axx applies. Without them, a client writes and reads keys and payloads as text, reads every record of the topic, and uses no consumer group.

Java client properties Axx accepts 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: Axx sizes batches by max.request.size, and publishes one event at a time.
• 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: they tune Java, and Axx has no counterpart for them; its text is 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.

Those that fail the step, since 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 other *.class or *.classes property.

Add it to a project with axx pack add kafka.

Settings

In axx.yaml, under packs.kafka:

Setting	Takes	Values	Default
lenientUnions	Accept Avro union values written without their {“branch type”: value} wrapper when exactly one branch fits; Avro’s JSON encoding needs the wrapper.	true, false	false
maxRecords	Records kept in memory per topic; older records are dropped and can no longer match.	a number	100000
timeout	How long a consumer assertion waits for a matching record, like 45s.	text	30s

Steps

kafka.service

Given the {word} kafka service with the following properties:
  | property | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Register a Kafka cluster. The first one registered in a scenario is the default for steps without on the {word} kafka service.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

Property	Takes	Default
brokers	the cluster’s brokers, a comma-separated list of host:port; ${env:..} and ${sys:..} are expanded	required
asyncapi	an AsyncAPI document, a file of the project or a URL, that the topics follow: every message a scenario sends there, and every message a check finds, is checked against it (with the asyncapi pack)

Any other property is ignored, with a warning.

Example:

Given the events kafka service with the following properties:
  | brokers | localhost:9092 |

kafka.client

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.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

Example:

Given the parcel-events kafka topic client

kafka.client.props

Given a(n) {word} kafka topic client[[ on the {word} kafka service]] with the following properties:
  | property | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Create a topic client configured with Java Kafka client properties: producer. rows for publishing, consumer. rows for assertions.

• For Avro, set the Confluent Avro serializer or deserializer and the schema.registry.url, as in the examples.
• A topic can have one client per service in a scenario.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

Property	Takes	Values	Default
<client>.bootstrap.servers	the brokers the client connects to, a comma-separated list; the service’s brokers by default
<client>.client.id	the name the client gives the brokers		axx
producer.key.serializer	how the producer writes keys: StringSerializer as text, ByteArraySerializer as the text’s bytes, KafkaAvroSerializer as an Avro string	org.apache.kafka.common.serialization.StringSerializer, org.apache.kafka.common.serialization.ByteArraySerializer, io.confluent.kafka.serializers.KafkaAvroSerializer	org.apache.kafka.common.serialization.StringSerializer
producer.value.serializer	how the producer writes payloads: StringSerializer and ByteArraySerializer as text, KafkaAvroSerializer as Avro, which publishing with a schema needs	org.apache.kafka.common.serialization.StringSerializer, org.apache.kafka.common.serialization.ByteArraySerializer, io.confluent.kafka.serializers.KafkaAvroSerializer	org.apache.kafka.common.serialization.StringSerializer
consumer.key.deserializer	how assertions read keys: StringDeserializer and ByteArrayDeserializer as text, KafkaAvroDeserializer as Avro	org.apache.kafka.common.serialization.StringDeserializer, org.apache.kafka.common.serialization.ByteArrayDeserializer, io.confluent.kafka.serializers.KafkaAvroDeserializer	org.apache.kafka.common.serialization.StringDeserializer
consumer.value.deserializer	how assertions read payloads: StringDeserializer and ByteArrayDeserializer as text, KafkaAvroDeserializer as Avro, whose properties are checked on Avro’s text form of the record	org.apache.kafka.common.serialization.StringDeserializer, org.apache.kafka.common.serialization.ByteArrayDeserializer, io.confluent.kafka.serializers.KafkaAvroDeserializer	org.apache.kafka.common.serialization.StringDeserializer
<client>.schema.registry.url	the Schema Registry’s URLs, a comma-separated list, which the Avro serializers need; user:password@ in a URL signs in with basic auth
<client>.basic.auth.credentials.source	where the registry’s basic auth credentials come from: the URL, basic.auth.user.info (USER_INFO), or the SASL username and password (SASL_INHERIT)	URL, USER_INFO, SASL_INHERIT	URL
<client>.basic.auth.user.info	user:password for the registry, with basic.auth.credentials.source=USER_INFO
<client>.schema.registry.basic.auth.user.info	the older name of basic.auth.user.info
<client>.bearer.auth.credentials.source	where the registry’s bearer token comes from: only STATIC_TOKEN, the bearer.auth.token	STATIC_TOKEN
<client>.bearer.auth.token	a bearer token for the registry
producer.auto.register.schemas	true registers the schema under its subject; false looks its ID up, and fails if it is not registered		true
producer.use.latest.version	true writes with the subject’s latest schema and its ID, with auto.register.schemas=false		false
producer.normalize.schemas	true has the registry normalize the schema when registering it or looking it up		false
producer.use.schema.id	the ID of the schema to write with, with auto.register.schemas=false
producer.key.subject.name.strategy	the subject of a key’s schema: <topic>-key (TopicNameStrategy); the record strategies fail, since a key is an Avro string	io.confluent.kafka.serializers.subject.TopicNameStrategy, io.confluent.kafka.serializers.subject.RecordNameStrategy, io.confluent.kafka.serializers.subject.TopicRecordNameStrategy	io.confluent.kafka.serializers.subject.TopicNameStrategy
producer.value.subject.name.strategy	the subject of a payload’s schema: <topic>-value (TopicNameStrategy), the record’s full name (RecordNameStrategy), or <topic>-<full name> (TopicRecordNameStrategy)	io.confluent.kafka.serializers.subject.TopicNameStrategy, io.confluent.kafka.serializers.subject.RecordNameStrategy, io.confluent.kafka.serializers.subject.TopicRecordNameStrategy	io.confluent.kafka.serializers.subject.TopicNameStrategy
<client>.schema.reflection	only false: reflection needs Java classes	false
consumer.specific.avro.reader	only false: Axx has no generated classes and reads generic records	false
<client>.schema.registry.ssl.<property>	an ssl. property of this table, for HTTPS to the Schema Registry
producer.acks	the acknowledgements a publish waits for: from every in-sync replica (all, -1), from none (0) or from the leader (1); 0 and 1 turn idempotence off unless enable.idempotence is set, as in Java	all, -1, 0, 1	all
producer.enable.idempotence	false turns idempotent publishing off; it is on by default with acks=all, and true needs acks=all
producer.compression.type	how the producer compresses what it sends	none, gzip, snappy, lz4, zstd	none
producer.linger.ms	how long the producer waits to fill a batch, in milliseconds, as in Java		5
producer.max.request.size	the largest batch the producer sends, in bytes
producer.buffer.memory	the most bytes of records the producer holds before they are sent
producer.delivery.timeout.ms	how long a publish may take, retries included, in milliseconds
producer.retries	how many times a failed publish is retried
producer.max.in.flight.requests.per.connection	how many publish requests may wait for an answer from a broker at once, with idempotence off
producer.partitioner.class	how records are spread over the partitions: by the key’s murmur2 hash (DefaultPartitioner), in turn (RoundRobinPartitioner) or a batch at a time (UniformStickyPartitioner); other classes fail the step	org.apache.kafka.clients.producer.internals.DefaultPartitioner, org.apache.kafka.clients.producer.RoundRobinPartitioner, org.apache.kafka.clients.producer.UniformStickyPartitioner	org.apache.kafka.clients.producer.internals.DefaultPartitioner
producer.transactional.id	nothing: it fails the step, since the publish steps run no transactions and a transactional producer cannot send
consumer.auto.offset.reset	the records assertions consider: every record of the topic (earliest), or those produced after the assertion starts (latest); none fails the step, since Axx reads without a consumer group	earliest, latest	earliest
consumer.isolation.level	read_committed leaves out the records of transactions that are open or were aborted	read_uncommitted, read_committed	read_uncommitted
consumer.allow.auto.create.topics	true lets reading a topic that does not exist create it; producers always may, as in Java		true
consumer.fetch.max.wait.ms	how long a broker may wait to fill a fetch, in milliseconds
consumer.fetch.min.bytes	the fewest bytes a broker answers a fetch with
consumer.fetch.max.bytes	the most bytes a broker answers a fetch with
consumer.max.partition.fetch.bytes	the most bytes of one partition a broker answers a fetch with
<client>.security.protocol	how the client connects: in plain text or over TLS (SSL), and signed in with SASL (SASL_) or not	PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL	PLAINTEXT
<client>.sasl.mechanism	how the client signs in with SASL; GSSAPI and OAUTHBEARER are not supported	PLAIN, SCRAM-SHA-256, SCRAM-SHA-512	PLAIN
<client>.sasl.jaas.config	the username and password the client signs in with, in a PlainLoginModule or ScramLoginModule entry; a SASL_ protocol needs it
<client>.ssl.truststore.location	the CA certificates the client trusts: a JKS, PKCS12 or PEM file, resolved against resources
<client>.ssl.truststore.password	the truststore’s password, which a JKS file may do without, as in Java
<client>.ssl.truststore.type	the truststore’s format, for a file Axx does not recognize by its content as JKS, PKCS12 or PEM	JKS, PKCS12, PEM	JKS
<client>.ssl.truststore.certificates	CA certificates the client trusts, in PEM, written in the cell
<client>.ssl.keystore.location	the client’s certificate and key, for mutual TLS: a JKS, PKCS12 or PEM file, resolved against resources
<client>.ssl.keystore.password	the keystore’s password
<client>.ssl.key.password	the private key’s password, for JKS key entries and encrypted PEM keys; the keystore’s password by default
<client>.ssl.keystore.type	the keystore’s format, for a file Axx does not recognize by its content as JKS, PKCS12 or PEM	JKS, PKCS12, PEM	JKS
<client>.ssl.keystore.key	the client’s private key, in PEM, written in the cell; with ssl.keystore.certificate.chain
<client>.ssl.keystore.certificate.chain	the client’s certificate chain, in PEM, written in the cell
<client>.ssl.endpoint.identification.algorithm	https checks the broker’s host name against its certificate; an empty value skips that check, and the certificate is still verified		https
<client>.ssl.protocol	the oldest TLS version the client accepts; TLS is TLSv1.2	TLS, TLSv1.2, TLSv1.3	TLSv1.2
<client>.ssl.enabled.protocols	the TLS versions the client may use, a comma-separated list of TLSv1.2 and TLSv1.3; older versions are left out
<client>.metadata.max.age.ms	how often the client refreshes what it knows of the cluster, in milliseconds; assertions refresh every 5 seconds without it
<client>.connections.max.idle.ms	how long a connection may stay idle before the client closes it, in milliseconds
<client>.socket.connection.setup.timeout.ms	how long connecting to a broker may take, in milliseconds
<client>.request.timeout.ms	how long a broker has to answer a request, in milliseconds
<client>.retry.backoff.ms	how long the client waits before it retries a request, in milliseconds
producer.<property>	any other Java producer property: those the pack’s description lists have no effect or fail the step, and the others are ignored, with a warning
consumer.<property>	any other Java consumer property: those the pack’s description lists have no effect or fail the step, and the others are ignored, with a warning

<client> is producer, for publishing, or consumer, for assertions: Axx applies the Java Kafka client property after the prefix to that client. Values are expanded (${env:..}, ${sys:..}); an empty value fails the step, and a row with neither prefix is ignored, with a warning.

Variants, the parts in [[...]] said or left out:

• 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:

Example:

Given a depot-scans kafka topic client with the following properties:
  | producer.value.serializer    | io.confluent.kafka.serializers.KafkaAvroSerializer |
  | producer.schema.registry.url | http://localhost:9081                              |
Given a parcel-events kafka topic client on the events kafka service with the following properties:
  | consumer.value.deserializer  | io.confluent.kafka.serializers.KafkaAvroDeserializer |
  | consumer.schema.registry.url | http://localhost:9081                                |

kafka.event

Given a(n)[[ {ordinal} ordered]] {word} kafka event[[ on {word} kafka service]]

Draft a new event of the topic: the steps that follow set its key, headers and payload, which 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 events the topic has still appends, with a warning.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101

Variants, the parts in [[...]] said or left out:

• 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

Example:

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

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.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101

Variants, the parts in [[...]] said or left out:

• 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

Example:

Given the depot-scans kafka event key is PX-4101
Given the 2nd ordered depot-scans kafka event key is PX-4102 on the events kafka service

kafka.event.headers

Given the[[ {ordinal} ordered]] {word} kafka event headers[[ on the {word} kafka service]] are:
  | header | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Set headers 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.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101

A row sets a header of the event: setting a header again replaces its value, and an empty cell sends the text null.

Variants, the parts in [[...]] said or left out:

• 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:

Example:

Given the depot-scans kafka event headers are:
  | X-Event-Type | ParcelScanned |
  | X-Depot      | Leipzig       |
Given the 1st ordered depot-scans kafka event headers on the events kafka service are:
  | X-Event-Type | ParcelScanned |

kafka.event.payload.resource

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.

• For Avro events, the file is Avro’s JSON encoding of the record, with unions as {"<branch>": 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.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101
{filepath}	a file of the project: a path relative to the feature file’s directory, the resources directories or axx.yaml’s directory (looked for in that order), or an absolute path, quoted when it has a space	seeds/parcels.yaml, kafka/scan-delivered.json

Variants, the parts in [[...]] said or left out:

• 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

Example:

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

Given the[[ {ordinal} ordered]] kafka event payload properties[[ on the {word} kafka service]] are:
  | JSONPath | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Set JSONPath properties of an event of the service’s first topic client, the first one created in the scenario.

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

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101

A row sets the property at the JSONPath to the value, always as a string; an empty cell sets JSON null.

Variants, the parts in [[...]] said or left out:

• 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:

Example:

Given the kafka event payload properties are:
  | $.scanId    | SC-4101-1 |
  | $.parcelRef | PX-4101   |
Given the 2nd ordered kafka event payload properties on the events kafka service are:
  | $.scanId    | SC-4101-2 |
  | $.parcelRef | PX-4101   |

kafka.event.properties

Given the {word} kafka event payload properties[[ on the {word} kafka service]] are:
  | JSONPath | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Set JSONPath properties of the topic’s first event. Every property must already exist in the payload: set it in the payload file first.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

A row sets the property at the JSONPath to the value, always as a string; an empty cell sets JSON null.

Variants, the parts in [[...]] said or left out:

• the {word} kafka event payload properties are:
• the {word} kafka event payload properties on the {word} kafka service are:

Example:

Given the depot-scans kafka event payload properties are:
  | $.scanId    | SC-4101-1 |
  | $.parcelRef | PX-4101   |
Given the depot-scans kafka event payload properties on the events kafka service are:
  | $.location | Leipzig |

kafka.event.properties.ordinal

Given the {ordinal} ordered {word} kafka event payload properties[[ on the {word} kafka service]] are:
  | JSONPath | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Set JSONPath properties of that event of the topic (1st is the first event created). Every property must already exist in the payload: set it in the payload file first.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101

A row sets the property at the JSONPath to the value, always as a string; an empty cell sets JSON null.

Variants, the parts in [[...]] said or left out:

• the {ordinal} ordered {word} kafka event payload properties are:
• the {ordinal} ordered {word} kafka event payload properties on the {word} kafka service are:

Example:

Given the 2nd ordered depot-scans kafka event payload properties are:
  | $.scanId    | SC-4101-2 |
  | $.parcelRef | PX-4101   |

Since 0.1.0.

kafka.event.property.null

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.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101

Variants, the parts in [[...]] said or left out:

• 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

Example:

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

When the[[ {ordinal} ordered]] {word} kafka event is published using schema {filepath}[[ on the {word} kafka service]]

Publish a drafted event as Confluent Avro, with an .avsc schema file.

• The payload, Avro’s JSON encoding of the record, is read with the schema; a payload that does not fit it fails the step with the JSONPath of the mismatch.
• The schema is registered, or looked up, in the Schema Registry under the subject of value.subject.name.strategy, <topic>-value by default.
• The record is written as magic byte 0, the schema ID and the Avro binary.
• The topic client needs producer.value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer and producer.schema.registry.url.
• 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.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101
{filepath}	a file of the project: a path relative to the feature file’s directory, the resources directories or axx.yaml’s directory (looked for in that order), or an absolute path, quoted when it has a space	seeds/parcels.yaml, kafka/scan-delivered.json

Variants, the parts in [[...]] said or left out:

• 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

Example:

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

When the[[ {ordinal} ordered]] {word} kafka event is published[[ on the {word} kafka service]]

Publish a drafted event as it is: its key, its headers and its payload’s text.

• The producer’s value serializer writes the payload: StringSerializer by default, or ByteArraySerializer.
• For Avro, use is published using schema: with KafkaAvroSerializer, this step fails.
• 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.

Parameter	Takes	For example
{ordinal}	a position, counting from 1; an optional ordinal left out is the first	1st, 2nd
{word}	one word, with no spaces	parcels, PX-4101

Variants, the parts in [[...]] said or left out:

• 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

Example:

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

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.
• The expectation joins those of the label (named {word}) in the scenario: one record of the topic must meet every one of them, 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; the failure lists the label’s expectations and the latest records, each with the reason it did not match.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

Variants, the parts in [[...]] said or left out:

• the {word} kafka event named {word} key is {word}
• the {word} kafka event named {word} key is {word} on the {word} kafka service

Example:

Then the parcel-events kafka event named registered key is PX-4101
Then the parcel-events kafka event named registered key is PX-4101 on the events kafka service

kafka.consumed.properties

Then the {word} kafka event named {word} payload properties[[ on the {word} kafka service]] are:
  | JSONPath | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Expect JSONPath properties of the label’s record payload.

• Values are typed: "text" is a string, null JSON null, 12 an integer, 1.5 a decimal, true and false booleans, {...} and [...] JSON; anything else is a string.
• Numbers must match in type: 2 does not equal 2.0.
• An empty cell fails the step: write null for JSON null, "" for an empty string.
• The expectation joins those of the label (named {word}) in the scenario: one record of the topic must meet every one of them, 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; the failure lists the label’s expectations and the latest records, each with the reason it did not match.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

A row expects the property at the JSONPath to have the value.

Variants, the parts in [[...]] said or left out:

• the {word} kafka event named {word} payload properties are:
• the {word} kafka event named {word} payload properties on the {word} kafka service are:

Example:

Then the parcel-events kafka event named registered payload properties are:
  | $.reference    | PX-4101  |
  | $.weightGrams  | 1200     |
  | $.serviceLevel | STANDARD |
Then the parcel-events kafka event named registered payload properties on the events kafka service are:
  | $.reference | PX-4101 |
  | $.source    | api     |

kafka.consumed.headers

Then the {word} kafka event named {word} headers[[ on the {word} kafka service]] are:
  | header | value |

The table may start with the row naming its columns, as above, or go straight to its rows.

Expect headers of the label’s record.

• Each header must occur exactly once, with exactly this value.
• The expectation joins those of the label (named {word}) in the scenario: one record of the topic must meet every one of them, 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; the failure lists the label’s expectations and the latest records, each with the reason it did not match.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

A row expects a header of the record to have the value; an empty cell fails the step.

Variants, the parts in [[...]] said or left out:

• the {word} kafka event named {word} headers are:
• the {word} kafka event named {word} headers on the {word} kafka service are:

Example:

Then the parcel-events kafka event named registered headers are:
  | X-Event-Type | ParcelRegistered |
Then the parcel-events kafka event named registered headers on the events kafka service are:
  | X-Event-Type | ParcelRegistered |

kafka.consumed.headers.match

Then the {word} kafka event named {word} headers match:
  | header | pattern |

The table may start with the row naming its columns, as above, or go straight to its rows.

Expect headers of the label’s record to match regular expressions.

• A header must have one distinct value: unlike the other steps, this one ignores repeated identical values of a header.
• The expectation joins those of the label (named {word}) in the scenario: one record of the topic must meet every one of them, 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; the failure lists the label’s expectations and the latest records, each with the reason it did not match.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

A row expects a header of the record to match a regular expression (Java syntax), over its whole value; an empty cell fails the step.

Example:

Then the parcel-events kafka event named registered headers match:
  | X-Event-Type | Parcel[A-Za-z]+ |

kafka.consumed.headers.match.service

Then the {word} kafka event named {word} headers on the {word} kafka service match:
  | header | pattern |

The table may start with the row naming its columns, as above, or go straight to its rows.

Expect headers of the label’s record, on a topic client of that Kafka service, to match regular expressions.

• A header must have one distinct value: unlike the other steps, this one ignores repeated identical values of a header.
• The expectation joins those of the label (named {word}) in the scenario: one record of the topic must meet every one of them, 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; the failure lists the label’s expectations and the latest records, each with the reason it did not match.

Parameter	Takes	For example
{word}	one word, with no spaces	parcels, PX-4101

A row expects a header of the record to match a regular expression (Java syntax), over its whole value; an empty cell fails the step.

Example:

Then the parcel-events kafka event named registered headers on the events kafka service match:
  | X-Event-Type | Parcel[A-Za-z]+ |

Since 0.1.0.