Kafka Plugin

plugins:
  - name: asyncapi
  - name: kafka
  - name: json

The plugin activates for AsyncAPI documents whose server declares protocol: kafka.

The plugin activates for AsyncAPI documents whose server declares protocol: kafka.

afka is the most complete transport. All four execution modes are available.

See AsyncAPI Execution Modes.

Drift reads the broker address from your AsyncAPI servers block:

servers:
  local:
    host: localhost:9092
    protocol: kafka
    description: Local Kafka broker

Override it per environment through plugin configuration, or per test through the test case. Test case values win, then plugin configuration, then the document.

The AsyncAPI channel address is the Kafka topic name:

channels:
  orderCreated:
    address: drift.examples.order-created

Drift matches captured messages against a correlation identifier carried in a Kafka record header. Declare its location in your message:

components:
  messages:
    orderCreated:
      correlationId:
        location: "$message.header#/correlation-id"

Declaring the location is required. The value is optional: Drift takes it from the AsyncAPI document, then from the correlation-id parameter in your test case, and generates a UUID when neither supplies one. Drift filters captured records to the one whose header matches, and ignores everything else on the topic.

Set the value yourself when a trigger hook or an expectation needs the same identifier. Where you do set it, give each operation a distinct value - a shared one lets a message from one test satisfy another.

When Drift subscribes, it records the current end of the topic log and reads forward from there. Messages published before the subscription are not captured, which is what makes the subscribe-then-trigger ordering reliable.

Drift polls for new records until a matching message arrives or the capture window closes.

parameters:
  timeout-ms: 5000

Lower poll-interval-ms reduces capture latency at the cost of more requests to the broker. The default suits local development and CI.

Drift does not create topics. Create every topic in your AsyncAPI document, and any probe topic, before the run.

Kafka's automatic topic creation fires only when a producer or consumer first touches a topic. That is too late for async-inject-capture, where Drift publishes to the input topic before any consumer has subscribed.

docker exec <kafka-container> kafka-topics \
  --bootstrap-server localhost:9092 \
  --create --if-not-exists \
  --topic drift.examples.order-created \
  --partitions 1 --replication-factor 1

A local broker is the recommended setup for development and CI. It is fast, repeatable, and nothing you publish affects anyone else.

services:
  kafka:
    image: confluentinc/cp-kafka:latest
    ports:
      - "9092:9092"
    environment:
      KAFKA_ENABLE_KRAFT: "yes"
      KAFKA_NODE_ID: "1"
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:9093"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1"
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "1"
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: "1"
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"

Running against a shared cluster is supported, but that cluster is yours to manage. Messages Drift publishes are real messages, and downstream consumers will process them.

The Kafka plugin speaks the Kafka wire protocol over raw TCP sockets. It does not support authentication or encryption in transit:

Point Drift at a broker that accepts plaintext connections, and treat contract verification as something you run against a local, disposable, or otherwise non-sensitive broker rather than a secured production cluster.