Skip to content

Kafka configuration

This guide covers how to configure Apache Kafka for integration testing with tomato.

Preset

The shortest setup is the kafka preset. tomato picks the host port, advertises it, and points the kafka resource at it:

containers:
  kafka:
    preset: kafka

resources:
  events:
    type: kafka
    container: kafka
    options:
      topics: [orders]

It is a single-node KRaft broker with these listeners:

Listener For Address
plain the host: tomato, an app run as a local process {{.kafka.host}}:{{.kafka.port.9092}}
plain other containers on tomato's network kafka:29092
AWS_MSK_IAM (auth: aws_msk_iam) the host {{.kafka.host}}:{{.kafka.port.9098}}
AWS_MSK_IAM (auth: aws_msk_iam) other containers kafka:29098

The kafka resource always uses the plain listener.

MSK IAM authentication

auth: aws_msk_iam adds a listener that speaks SASL AWS_MSK_IAM, the mechanism aws-msk-iam-auth uses against MSK. Point the app at it with the same client settings it uses in production, SASL_PLAINTEXT in place of SASL_SSL:

containers:
  kafka:
    preset: kafka
    auth: aws_msk_iam

resources:
  aws:
    type: aws
    options:
      role_arn: arn:aws:iam::000000000000:role/my-service

app:
  command: java -jar build/libs/my-service.jar
  env:
    SPRING_KAFKA_BOOTSTRAP_SERVERS: "{{.kafka.host}}:{{.kafka.port.9098}}"
    SPRING_KAFKA_PROPERTIES_SECURITY_PROTOCOL: SASL_PLAINTEXT
    SPRING_KAFKA_PROPERTIES_SASL_MECHANISM: AWS_MSK_IAM
    SPRING_KAFKA_PROPERTIES_SASL_JAAS_CONFIG: "software.amazon.msk.auth.iam.IAMLoginModule required;"
    SPRING_KAFKA_PROPERTIES_SASL_CLIENT_CALLBACK_HANDLER_CLASS: software.amazon.msk.auth.iam.IAMClientCallbackHandler

The listener checks who the client is, not its signature. With an aws resource in the config it lets in only the role sessions that resource's STS issues, and tells any other identity Access denied, as MSK does for a role without the kafka-cluster policy. So an app that ends up with the wrong credentials (its SDK could not reach STS and fell back to another identity, say) fails here the way it fails against MSK. Without an aws resource, any identity gets in.

The image

The preset runs the stock apache/kafka:3.9.1. For auth: aws_msk_iam, tomato copies the AWS_MSK_IAM server it carries in its binary into the container's /opt/kafka/libs before the broker starts, so there is no extra image to pull or build. Set image: on the entry to run another apache/kafka tag; the plugin is copied into it the same way.

Manual setup without the preset

Reach for this only when you need a broker the preset does not cover — a different distribution, or a Zookeeper-based cluster to match production. Otherwise use preset: kafka above; it is what tomato's own integration suite runs.

Why a broker needs extra configuration

Kafka's advertised listeners are fixed at startup, before Docker assigns a dynamic host port, so a client on the host cannot discover the mapped port. The way around it is a fixed port mapping for the external listener plus a second listener for traffic between containers. The preset does this for you.

Container setup

Zookeeper with Kafka and dual listeners:

containers:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.1
    env:
      ZOOKEEPER_CLIENT_PORT: "2181"
      ZOOKEEPER_TICK_TIME: "2000"
    ports:
      - "2181/tcp"
    wait_for:
      type: port
      target: "2181/tcp"
      timeout: 30s

  kafka:
    image: confluentinc/cp-kafka:7.6.1
    depends_on:
      - zookeeper
    env:
      KAFKA_BROKER_ID: "1"
      KAFKA_ZOOKEEPER_CONNECT: "{{.zookeeper.host}}:{{.zookeeper.port.2181}}"
      # Dual listeners: internal (kafka:29092) and external (localhost:9092)
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1"
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
    ports:
      # Fixed port for external access (host connects here)
      - "9092:9092"
      # Internal port for container-to-container (dynamic is fine)
      - "29092/tcp"
    wait_for:
      type: port
      target: "9092/tcp"
      timeout: 60s

Key configuration explained

Dual listeners

Kafka is configured with two listeners:

Listener Address Purpose
PLAINTEXT kafka:29092 Internal communication between containers
PLAINTEXT_HOST localhost:9092 External access from host machine (tests)

Fixed port mapping

The external port uses fixed mapping (9092:9092) so the advertised listener (localhost:9092) matches the actual port:

ports:
  - "9092:9092"    # Fixed: host:container
  - "29092/tcp"    # Dynamic: just container port

Environment variable templates

Tomato supports templates in container environment variables:

Template Resolves To
{{.container_name.host}} Container's DNS name (for inter-container communication)
{{.container_name.port.XXXX}} Container's internal port (not mapped host port)

For Zookeeper connection: 

KAFKA_ZOOKEEPER_CONNECT: "{{.zookeeper.host}}:{{.zookeeper.port.2181}}"

This resolves to zookeeper:2181 (container DNS name + internal port).

Resource configuration

Configure the Kafka resource to connect to the container:

resources:
  events:
    type: kafka
    container: kafka
    options:
      topics:
        - orders
        - events
        - notifications
      partitions: 1
      replication_factor: 1
      reset_strategy: delete_recreate

Resource options

Option Type Description
topics list Topics to create/manage
partitions int Number of partitions per topic (default: 1)
replication_factor int Replication factor (default: 1)
reset_strategy string How to reset between scenarios

Reset strategies

Strategy Description
delete_recreate Delete and recreate topics (recommended)
none No reset between scenarios

Complete example

A complete tomato.yml with Kafka, using the preset:

version: 2

settings:
  timeout: 5m
  fail_fast: false
  output: pretty

containers:
  kafka:
    preset: kafka

resources:
  events:
    type: kafka
    container: kafka
    options:
      topics:
        - user-events
        - order-events
      partitions: 1
      replication_factor: 1
      reset_strategy: delete_recreate

features:
  paths:
    - ./features

Writing Kafka tests

Once configured, you can write Gherkin scenarios to test Kafka:

Feature: Event Processing

  Scenario: Order events are published correctly
    Given "events" creates topic "order-events"
    And "events" consumes from "order-events"
    When "events" publishes json to "order-events" with key "order-123":
      """
      {"event": "order.created", "orderId": "123", "amount": 99.99}
      """
    Then "events" receives from "order-events" with key "order-123" within "10s"
    And "events" last message contains:
      """
      order.created
      """

See Kafka Steps for the complete list of available steps.

Avro and Schema Registry

Tomato speaks the Confluent wire format: a magic byte, a 4-byte schema id, then the Avro body. That is what KafkaAvroSerializer and KafkaAvroDeserializer read and write, so a JVM app under test sees ordinary Avro messages. Feature files stay plain JSON.

Add a schema registry

containers:
  schema-registry:
    image: confluentinc/cp-schema-registry:7.6.1
    depends_on:
      - kafka
    env:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:29092
    ports:
      - "8081/tcp"
    wait_for:
      type: http
      path: /subjects
      target: "8081"
      timeout: 90s

resources:
  kafka:
    type: kafka
    container: kafka
    options:
      topics: [orders]
      schema_registry:
        container: schema-registry     # or: url: http://localhost:8081
        subjects:                      # optional, topic -> subject
          payments: com.example.Payment

app:
  env:
    SPRING_KAFKA_PROPERTIES_SCHEMA_REGISTRY_URL: "http://{{.schema-registry.host}}:{{.schema-registry.port.8081}}"
Option Description
schema_registry.url Registry base URL
schema_registry.container Container running the registry; tomato resolves host and mapped port
schema_registry.port Registry port inside the container (default 8081)
schema_registry.subjects Map a topic to a subject. Default is <topic>-value (TopicNameStrategy)

Writing Avro tests

Scenario: Order service emits an OrderCreated event
  Given "kafka" registers schema for subject "orders-value" from file "schemas/order.avsc"
  And "kafka" consumes from "orders"
  When "api" sends "POST" to "/orders" with json:
    """
    {"id": "order-1", "amount": 4200}
    """
  Then "kafka" receives avro from "orders" within "10s":
    """
    {"id": "order-1", "status": "CREATED"}
    """
  And "kafka" last message avro matches:
    """
    {"id": "order-1", "amount": 4200, "status": "CREATED", "note": null}
    """
  • publishes avro to "topic" encodes the JSON with the latest schema of the topic's value subject. Register it first, or let the app register it.
  • receives avro from ... within waits for a message whose decoded JSON contains the given fields, so other messages on the topic don't break the test. last message avro matches compares the whole record exactly.
  • Nullable fields (["null", "string"]) are written as plain JSON values ("rush" or null), not Avro-JSON's {"string": "rush"}.
  • Keys are sent and asserted as strings. Avro keys, schema references and Protobuf/JSON Schema subjects are not supported yet.

AWS Glue Schema Registry

An app that uses the AWS Glue Schema Registry writes and reads a different wire format, and resolves schemas by calling Glue through the AWS SDK. With format: glue, tomato speaks that format and serves the Glue API itself, so the app fetches the schema the scenario registered and no AWS account is involved:

resources:
  events:
    type: kafka
    container: kafka
    options:
      topics: [trade-events]
      schema_registry:
        format: glue
        registry_name: portfolio   # default: default-registry
        region: eu-central-1       # in the schema ARNs; default us-east-1
        compression: zlib          # how tomato writes values: none (default) or zlib
        port: 9994                 # of the Glue endpoint; default: a free port
Option Default Description
format confluent glue for the AWS Glue Schema Registry
registry_name default-registry The registry schemas are registered in and looked up from
region us-east-1 Region in the schema ARNs the API returns
compression none zlib compresses the values tomato publishes. Values the app publishes are read either way
port free port Port of the Glue endpoint

Wire format. [0x03][0x00 none | 0x05 zlib][16-byte schema version UUID][Avro binary].

The app is pointed at it. The resource sets AWS_ENDPOINT_URL_GLUE in the app's environment, so the AWS SDK sends Glue calls to tomato. Pair it with the aws resource for the credentials those calls are signed with; tomato does not verify signatures. The endpoint answers GetSchemaVersion, GetSchemaByDefinition, CreateSchema, RegisterSchemaVersion and GetSchema, the operations the AWS Glue SerDe calls. Anything else gets an UnsupportedOperationException that names it.

Subjects are schema names. The steps are the same as for the Confluent registry; a topic's schema is named after the topic (the SerDe's default), unless subjects maps it:

Given "events" registers schema for subject "trade-events" from file "schemas/trade.avsc"
When "events" publishes avro to "trade-events" with key "user-1":
  """
  {"tradeId": "t-1", "quantity": "10"}
  """

Registered schemas last for the whole run, as in a real registry: the app caches schemas by version id. Registering a definition that is already a version of the schema, in any layout, returns that version.

Troubleshooting

Connection refused

If tests fail with "connection refused":

  1. Verify Kafka container is healthy: docker ps
  2. Check the broker's port is mapped: docker port <container_id>
  3. With the manual setup, make sure nothing else holds port 9092: lsof -i :9092

Consumer not receiving messages

  1. Ensure consumes from is called before publishing
  2. The consumer starts at OffsetNewest - only messages published after consumer starts are received
  3. Verify topic exists with topic "name" exists

Port already in use

The manual setup pins the external listener to 9092:9092, so a second Kafka on that port clashes. The preset picks a free host port and is unaffected.

# Kill any process using port 9092
lsof -ti:9092 | xargs kill -9