Skip to main content

Kafka Testing

Karate tests Kafka through the commercial kafka extension: produce and consume messages from ordinary feature files, assert on keys, values and headers with match, and mix Kafka steps with HTTP calls in one scenario — the common shape for "call the API, then check what landed on the topic". JSON works with zero setup; Avro and Protobuf are supported through registered schemas.

What you need
  • The karate-async-<version>.jar — the engine build that carries the async protocols. Download it from the karate-addons releases.
  • A license with the kafka entitlement, at .karate/karate.lic in your project (or the KARATE_LICENSE_TEXT environment variable).
  • JDK 21+.

Activate the extension​

Cluster coordinates are set once, in karate-boot.js at the project root — every producer and consumer in the suite is built from this config:

karate-boot.js
var kafka = boot.ext('kafka');
kafka.bootstrap = boot.sysprop('kafka.bootstrap', '127.0.0.1:29092');

// Optional — set it when your topics use Confluent Schema Registry serialization.
kafka.schemaRegistry = boot.sysprop('kafka.schemaRegistry', 'http://localhost:8081');

// Any other Kafka client property goes here verbatim — the escape hatch for SASL, compression, timeouts.
// kafka.props = { 'compression.type': 'gzip' };

boot.sysprop(name, default) reads a system property, so CI can re-point the suite without editing anything: java -Dkafka.bootstrap=broker.internal:9092 -jar karate-async-<version>.jar checks (the -D must come before -jar — that is where the JVM reads it).

Migrating from 0.x

configure kafka = ... is removed and now throws an error. Cluster settings live on the extension in karate-boot.js, as above — the four keys are bootstrap, schemaRegistry, ssl and props.

Write a check​

A channel is created with karate.channel('kafka'); from it you build consumers and producers. Start the consumer before you produce — each consumer joins with a fresh group id and reads from the latest offset, so it only sees records produced after start(). Consumers and producers are closed automatically at the end of each scenario.

checks/kafka.feature
Feature: Kafka checks

  Background:
    * def channel = karate.channel('kafka')

  Scenario: JSON round trip
    * def consumer = channel.consumer()
    * consumer.topic = 'json-topic'
    * consumer.count = 1
    * consumer.start()

    * def producer = channel.producer()
    * producer.topic = 'json-topic'
    * producer.key = 'k1'
    * producer.value = { message: 'hello', n: 42 }
    * producer.send()

    * def record = consumer.pop()
    * match record.key == 'k1'
    * match record.value == { message: 'hello', n: 42 }

  Scenario: headers travel with the record
    * def consumer = channel.consumer()
    * consumer.topic = 'json-topic'
    * consumer.start()

    * def producer = channel.producer()
    * producer.topic = 'json-topic'
    * producer.headers = { source: 'karate', tenant: 'acme' }
    * producer.value = { message: 'with headers' }
    * producer.send()

    * def record = consumer.pop()
    * match record.headers contains { source: 'karate', tenant: 'acme' }

Run it: java -jar karate-async-<version>.jar checks — the HTML report lands in target/karate-reports/.

Each record that comes back is { key, value, headers, offset }.

Configuration reference​

Boot-level keys (on boot.ext('kafka')):

KeyMeans
bootstrapthe bootstrap.servers list
schemaRegistrySchema Registry URL — optional; set it when your topics use registry serialization
sslTLS settings — see TLS and SASL
propsany Kafka client property, verbatim; merged last, so it overrides

Consumer keys:

KeyMeans
topicthe topic to subscribe to
counthow many records collect() waits for (default 1)
timeoutwait limit in milliseconds
filtera JS function — keep only the records it returns true for, e.g. x => x.key != 'skip-me'
schemaa registered schema name — see Avro and Protobuf
start()begin consuming — call this before producing
pop()take the next record
collect()take all records received (waits for count)

Producer keys:

KeyMeans
topicthe topic to send to
key, valuethe record — value can be JSON of any shape
headersa map of record headers
schemaa registered schema name
send()send the record

Avro and Protobuf​

Register a schema on the channel, then name it on the consumer and producer:

  Scenario: Avro through the Schema Registry
    * channel.register({ name: 'hello', path: '/hello.avsc' })

    * def consumer = channel.consumer()
    * consumer.topic = 'avro-topic'
    * consumer.schema = 'hello'
    * consumer.start()

    * def producer = channel.producer()
    * producer.topic = 'avro-topic'
    * producer.schema = 'hello'
    * producer.value = { message: 'hello', status: 'NEW', info: { first: 1, second: true } }
    * producer.send()

    * match consumer.pop().value == { message: 'hello', status: 'NEW', info: { first: 1, second: true } }
  • .avsc is Avro. With kafka.schemaRegistry set, records go through Confluent Schema Registry serialization; without it, the registered schema is used to read and write raw Avro bytes — no registry needed.
  • .proto is Protobuf, sent as bytes and decoded with the registered descriptor — no registry needed: channel.register({ name: 'hello-proto', path: '/hello.proto', message: 'Hello', roots: ['/'] }).
  • A custom wire format can be registered as a codec: channel.register({ name: 'mine', codec: myCodec }), where myCodec implements the WireCodec interface (encode/decode).

TLS and SASL​

karate-boot.js
var kafka = boot.ext('kafka');
kafka.ssl = {
  protocol: 'SSL',
  truststore: '/ssl/client.truststore.p12', truststorePassword: 'secret',
  keystore:   '/ssl/client.keystore.p12',   keystorePassword:   'secret', keyPassword: 'secret'
};
// PKCS12 keystores need this — the Kafka client assumes JKS otherwise
kafka.props = { 'ssl.truststore.type': 'PKCS12', 'ssl.keystore.type': 'PKCS12' };

For SASL, put the usual sasl.* client properties in kafka.props.

Path references​

A leading / always means the project root, never the file system root — so /hello.avsc is portable across machines and CI. For a real host path (a mounted secret, /etc/ssl/...), use file:/....

Runnable example​

A complete, CI-verified project — docker-compose broker + Schema Registry, schemas, checks and boot file — is at karate-agent-examples/kafka, with its live report. If a snippet on this page ever disagrees with that kit, the kit is the one gated by CI.

Part of Karate Agent

The karate-async jar is a build of Karate Agent — the AI-native engine that adds a served console, API coverage and requirements traceability over these same feature files. An event on a topic counts as coverage evidence like any HTTP call. See the enterprise overview.

See Also​

On This Page