Test Kafka Consumer Contracts Step by Step

TL;DR

Model the exact event your consumer promises to accept, test serialization first, then publish through a containerized broker and assert the business outcome. Add negative, duplicate, and failure-path cases so a passing suite proves compatibility rather than mere message delivery.

Key Takeaways

• Define the contract at the consumer boundary, including topic, key, headers, schema, and behavior.
• Keep fast serialization tests separate from broker-backed integration tests.
• Use Testcontainers so tests exercise a real Kafka-compatible broker without shared infrastructure.
• Verify observable business outcomes and committed offsets, not private listener methods.
• Test missing fields, unknown fields, invalid enums, duplicates, and malformed payloads explicitly.
• Run compatibility checks in CI before producers publish breaking schemas.

To test Kafka consumer contracts step by step, define what crosses the consumer boundary, prove that valid events deserialize, publish them through a real broker, and assert the consumer's observable result. A contract test is complete only when it also proves how incompatible, malformed, and duplicate events are handled.

This tutorial builds a small Java 21 consumer test suite with JUnit 5, Apache Kafka clients, Jackson, and Testcontainers. It complements the Event-Driven API Testing Complete Guide, which explains the wider strategy for topics, schemas, delivery guarantees, and observability.

You will keep the example framework-light so the contract remains visible. The same test layers apply to Spring Kafka, Quarkus, Micronaut, or a custom poll loop.

What You Will Build

You will build a consumer that reads orders.created.v1 events and records accepted orders. The finished suite will:

• Define the topic, key, headers, JSON shape, and semantic rules.
• Verify serialization and deserialization without starting Kafka.
• Start an isolated Kafka-compatible broker with Testcontainers.
• Publish an event and verify the consumer's business result.
• Reject incompatible payloads without silently advancing business state.
• Prove duplicate delivery is idempotent.
• Produce useful diagnostics in CI.

The example deliberately separates transport success from business success. send() succeeding proves that Kafka stored a record. It does not prove that the consumer understood or processed it.

Prerequisites

Use JDK 21, Maven 3.9 or newer, and Docker Engine or a Docker-compatible Testcontainers runtime. Confirm them first:

java -version
mvn -version
docker version

Create an empty Maven project. The examples use Java records and a current Testcontainers 2.x dependency line. Pin the newest compatible patch available in your dependency update process rather than using floating versions.

<project xmlns="http://maven.apache.org/POM/4.0.0">
  <modelVersion>4.0.0</modelVersion>
  <groupId>dev.qajobfit</groupId>
  <artifactId>kafka-consumer-contract</artifactId>
  <version>1.0-SNAPSHOT</version>
  <properties>
    <maven.compiler.release>21</maven.compiler.release>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <junit.version>5.12.2</junit.version>
    <testcontainers.version>2.0.2</testcontainers.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>4.1.1</version>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
      <version>2.20.1</version>
    </dependency>
    <dependency>
      <groupId>org.junit.jupiter</groupId>
      <artifactId>junit-jupiter</artifactId>
      <version>${junit.version}</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.testcontainers</groupId>
      <artifactId>testcontainers-kafka</artifactId>
      <version>${testcontainers.version}</version>
      <scope>test</scope>
    </dependency>
  </dependencies>
  <build><plugins><plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-surefire-plugin</artifactId>
    <version>3.5.3</version>
    <configuration><useModulePath>false</useModulePath></configuration>
  </plugin></plugins></build>
</project>

Verification: Run mvn -q test. Maven should exit with code 0, even though no tests exist yet. If Docker is remote, configure Testcontainers according to that runtime before continuing.

Step 1: Write the Consumer Contract Before the Test

A consumer contract is more than a JSON example. Record every input assumption and every observable outcome. For this tutorial, the contract is:

Save a representative fixture as src/test/resources/contracts/order-created-v1.json:

{
  "eventId": "evt-1001",
  "orderId": "ord-501",
  "customerId": "cus-42",
  "total": 79.95,
  "currency": "USD",
  "occurredAt": "2026-07-15T08:30:00Z"
}

Fixtures are examples, not the entire contract. The table supplies semantic expectations that JSON alone cannot express. Put this decision record beside the consumer code so producer and consumer teams can review it.

Verification: Compare the fixture with the table field by field. Confirm the key is ord-501, the event type header is required, and the total is nonnegative. A reviewer should be able to predict every test that follows.

Step 2: Implement Strict Required Fields and Tolerant Evolution

Create src/main/java/dev/qajobfit/OrderCreated.java:

package dev.qajobfit;

import java.math.BigDecimal;
import java.time.Instant;

public record OrderCreated(
    String eventId, String orderId, String customerId,
    BigDecimal total, String currency, Instant occurredAt) {

  public OrderCreated {
    if (eventId == null || eventId.isBlank()) throw new IllegalArgumentException("eventId is required");
    if (orderId == null || orderId.isBlank()) throw new IllegalArgumentException("orderId is required");
    if (customerId == null || customerId.isBlank()) throw new IllegalArgumentException("customerId is required");
    if (total == null || total.signum() < 0) throw new IllegalArgumentException("total must be nonnegative");
    if (currency == null || currency.length() != 3) throw new IllegalArgumentException("currency must have 3 letters");
    if (occurredAt == null) throw new IllegalArgumentException("occurredAt is required");
  }
}

Create src/main/java/dev/qajobfit/Json.java:

package dev.qajobfit;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.json.JsonMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;

public final class Json {
  public static final ObjectMapper MAPPER = JsonMapper.builder()
      .addModule(new JavaTimeModule())
      .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
      .build();
  private Json() {}
}

Add jackson-datatype-jsr310 at the same Jackson version to the Maven dependencies. Ignoring unknown properties is intentional: adding an optional producer field should not break an older consumer. Required constructor checks remain strict.

Verification: Compile with mvn -q test. The build should pass. Temporarily remove eventId from the fixture in a small unit test and confirm deserialization throws an exception whose cause reports eventId is required.

Step 3: Test the Contract Without Kafka

Create src/test/java/dev/qajobfit/OrderCreatedContractTest.java:

package dev.qajobfit;

import static org.junit.jupiter.api.Assertions.*;

import com.fasterxml.jackson.databind.JsonNode;
import java.nio.file.Files;
import java.nio.file.Path;
import org.junit.jupiter.api.Test;

class OrderCreatedContractTest {
  private static String fixture() throws Exception {
    return Files.readString(Path.of("src/test/resources/contracts/order-created-v1.json"));
  }

  @Test void readsThePublishedContract() throws Exception {
    OrderCreated event = Json.MAPPER.readValue(fixture(), OrderCreated.class);
    assertEquals("ord-501", event.orderId());
    assertEquals("USD", event.currency());
    assertEquals("79.95", event.total().toPlainString());
  }

  @Test void acceptsAnAdditiveField() throws Exception {
    JsonNode original = Json.MAPPER.readTree(fixture());
    String evolved = ((com.fasterxml.jackson.databind.node.ObjectNode) original)
        .put("salesChannel", "mobile").toString();
    assertDoesNotThrow(() -> Json.MAPPER.readValue(evolved, OrderCreated.class));
  }

  @Test void rejectsMissingRequiredField() throws Exception {
    JsonNode original = Json.MAPPER.readTree(fixture());
    String broken = ((com.fasterxml.jackson.databind.node.ObjectNode) original)
        .remove("eventId").toString();
    Exception error = assertThrows(Exception.class,
        () -> Json.MAPPER.readValue(broken, OrderCreated.class));
    assertTrue(error.getMessage().contains("eventId is required"));
  }
}

These are fast contract boundary tests. They catch field renames, type changes, required-field removal, and accidental strictness toward new fields. They do not prove topic configuration, headers, polling, or offset behavior.

Verification: Run mvn -q -Dtest=OrderCreatedContractTest test. Expect three passing tests. Change total to "seventy"; readsThePublishedContract should fail during mapping, proving the suite detects a type break.

Step 4: Build a Small Observable Consumer

Create src/main/java/dev/qajobfit/OrderConsumer.java:

package dev.qajobfit;

import java.time.Duration;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public final class OrderConsumer implements Runnable, AutoCloseable {
  private final KafkaConsumer<String, String> kafka;
  private final Set<String> processedIds = ConcurrentHashMap.newKeySet();
  private final CopyOnWriteArrayList<OrderCreated> accepted = new CopyOnWriteArrayList<>();
  private final CopyOnWriteArrayList<String> failures = new CopyOnWriteArrayList<>();
  private volatile boolean running = true;

  public OrderConsumer(KafkaConsumer<String, String> kafka) { this.kafka = kafka; }
  public java.util.List<OrderCreated> accepted() { return java.util.List.copyOf(accepted); }
  public java.util.List<String> failures() { return java.util.List.copyOf(failures); }

  @Override public void run() {
    try {
      while (running) {
        for (ConsumerRecord<String, String> record : kafka.poll(Duration.ofMillis(200))) {
          try {
            String type = new String(record.headers().lastHeader("event-type").value());
            if (!"OrderCreated".equals(type)) throw new IllegalArgumentException("wrong event-type");
            OrderCreated event = Json.MAPPER.readValue(record.value(), OrderCreated.class);
            if (!record.key().equals(event.orderId())) throw new IllegalArgumentException("key must equal orderId");
            if (processedIds.add(event.eventId())) accepted.add(event);
          } catch (Exception error) {
            failures.add(error.getMessage());
          }
        }
      }
    } finally { kafka.close(); }
  }

  @Override public void close() { running = false; kafka.wakeup(); }
}

In production, route failures to a durable dead-letter topic rather than an in-memory list. This seam exists only to make the tutorial's outcome observable. Notice that the test will assert accepted business events, not call run() internals directly.

The header access above assumes the required header exists. A missing header causes a caught failure, which matches the declared contract. Production code should record a structured reason and the original topic, partition, and offset.

Verification: Run mvn -q test. Compilation proves the client API is wired correctly. Review that processedIds.add and accepted.add form one logical decision; a real database implementation should enforce uniqueness atomically.

Step 5: Test Kafka Consumer Contracts Step by Step Through a Broker

Create src/test/java/dev/qajobfit/OrderConsumerIntegrationTest.java. This test starts one broker per class, creates a unique group, publishes a record, and waits for the business effect.

package dev.qajobfit;

import static org.junit.jupiter.api.Assertions.*;

import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.Properties;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.*;
import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.utility.DockerImageName;

class OrderConsumerIntegrationTest {
  static final String TOPIC = "orders.created.v1";
  static final KafkaContainer KAFKA = new KafkaContainer(
      DockerImageName.parse("apache/kafka-native:4.1.1"));
  OrderConsumer consumer;
  Thread thread;

  @BeforeAll static void startBroker() { KAFKA.start(); }
  @AfterAll static void stopBroker() { KAFKA.stop(); }

  @BeforeEach void startConsumer() {
    Properties p = new Properties();
    p.put("bootstrap.servers", KAFKA.getBootstrapServers());
    p.put("group.id", "contract-" + java.util.UUID.randomUUID());
    p.put("auto.offset.reset", "earliest");
    p.put("enable.auto.commit", "false");
    KafkaConsumer<String,String> kafka = new KafkaConsumer<>(p, new StringDeserializer(), new StringDeserializer());
    kafka.subscribe(java.util.List.of(TOPIC));
    consumer = new OrderConsumer(kafka);
    thread = Thread.ofPlatform().start(consumer);
  }

  @AfterEach void stopConsumer() throws Exception { consumer.close(); thread.join(3000); }

  @Test void consumesAContractCompatibleEvent() throws Exception {
    String json = Files.readString(Path.of("src/test/resources/contracts/order-created-v1.json"));
    publish("ord-501", json, "OrderCreated");
    await(() -> consumer.accepted().size() == 1, Duration.ofSeconds(10));
    assertEquals("ord-501", consumer.accepted().getFirst().orderId());
    assertTrue(consumer.failures().isEmpty());
  }

  static void publish(String key, String value, String type) throws Exception {
    Properties p = new Properties();
    p.put("bootstrap.servers", KAFKA.getBootstrapServers());
    try (KafkaProducer<String,String> producer = new KafkaProducer<>(p, new StringSerializer(), new StringSerializer())) {
      ProducerRecord<String,String> record = new ProducerRecord<>(TOPIC, null, key, value,
          java.util.List.of(new RecordHeader("event-type", type.getBytes(java.nio.charset.StandardCharsets.UTF_8))));
      producer.send(record).get();
    }
  }

  static void await(java.util.function.BooleanSupplier condition, Duration timeout) throws Exception {
    long end = System.nanoTime() + timeout.toNanos();
    while (!condition.getAsBoolean() && System.nanoTime() < end) Thread.sleep(50);
    assertTrue(condition.getAsBoolean(), "condition was not met within " + timeout);
  }
}

A bounded polling assertion is better than a fixed multi-second sleep. It returns quickly on success and gives the consumer enough time on a loaded CI runner. A unique group prevents state leakage between tests.

Verification: Run mvn -q -Dtest=OrderConsumerIntegrationTest test. Expect one passing test and container startup logs. If it times out, print consumer.failures() before assuming Kafka delivery failed.

Step 6: Add Breaking and Malformed Event Tests

Add these methods to OrderConsumerIntegrationTest:

@Test void rejectsATypeBreakingPayload() throws Exception {
  String broken = """
      {"eventId":"evt-bad","orderId":"ord-bad","customerId":"cus-42",
       "total":"unknown","currency":"USD","occurredAt":"2026-07-15T08:30:00Z"}
      """;
  publish("ord-bad", broken, "OrderCreated");
  await(() -> consumer.failures().size() == 1, Duration.ofSeconds(10));
  assertTrue(consumer.accepted().isEmpty());
}

@Test void rejectsTheWrongRecordKey() throws Exception {
  String json = Files.readString(Path.of("src/test/resources/contracts/order-created-v1.json"));
  publish("different-order", json, "OrderCreated");
  await(() -> consumer.failures().size() == 1, Duration.ofSeconds(10));
  assertTrue(consumer.failures().getFirst().contains("key must equal orderId"));
}

@Test void rejectsTheWrongEventType() throws Exception {
  String json = Files.readString(Path.of("src/test/resources/contracts/order-created-v1.json"));
  publish("ord-501", json, "OrderCancelled");
  await(() -> consumer.failures().size() == 1, Duration.ofSeconds(10));
  assertTrue(consumer.accepted().isEmpty());
}

These cases protect different contract surfaces. The first catches schema incompatibility, the second catches partition-key drift, and the third catches routing errors. Add missing-header, invalid timestamp, negative total, truncated JSON, and unsupported currency cases when those risks exist in your domain.

Do not assert an exact Jackson exception string. Library patches can improve wording. Assert the stable business classification or a meaningful substring, and preserve the full exception in logs.

Verification: Run the integration test class again. Expect four tests to pass. Then change the consumer to ignore the record key. rejectsTheWrongRecordKey must fail, demonstrating that the test detects weakened contract enforcement.

Step 7: Prove Duplicate Delivery Is Safe

Kafka commonly provides at-least-once processing. Rebalances, retries, or a crash after the business write but before offset commit can deliver a record again. Add this test:

@Test void processesTheSameEventIdOnlyOnce() throws Exception {
  String json = Files.readString(Path.of("src/test/resources/contracts/order-created-v1.json"));
  publish("ord-501", json, "OrderCreated");
  publish("ord-501", json, "OrderCreated");
  await(() -> consumer.accepted().size() == 1, Duration.ofSeconds(10));
  Thread.sleep(300);
  assertEquals(1, consumer.accepted().size());
  assertTrue(consumer.failures().isEmpty());
}

The short sleep here is not the primary success mechanism. It is a stability window after the first observed result, used to expose a late second write. For a production consumer, assert a unique database constraint or idempotency ledger rather than an in-memory set.

Also test two different event IDs for the same order if updates are valid. Idempotency should normally use an immutable event ID, not only the aggregate ID, because separate events for one order can both be legitimate.

Verification: Run the class and expect five passing tests. Remove processedIds.add from the conditional. The duplicate test should report two accepted items and fail. Restore the guard after proving the test's sensitivity.

Step 8: Add the Contract Suite to CI

Run fast mapping tests on every change and broker tests before merge. A GitHub Actions job can use the runner's Docker service directly:

name: kafka-consumer-contract
on:
  pull_request:
  push:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with:
          distribution: temurin
          java-version: '21'
          cache: maven
      - name: Run consumer contract tests
        run: mvn --batch-mode --no-transfer-progress verify
      - name: Upload test reports
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: surefire-reports
          path: target/surefire-reports/

For schema-registry formats, add a compatibility gate before broker tests. The step-by-step Avro schema compatibility guide shows how to prevent incompatible schemas from merging. Broker tests still matter because registry compatibility alone does not validate headers, keys, subscriptions, or business behavior.

Keep container images and dependencies pinned. Let an automated dependency process propose upgrades, then run the full contract suite against each upgrade. Cache Maven artifacts, not broker state, so tests remain isolated.

Verification: Open a pull request with a deliberate fixture break, such as renaming orderId to id. The job should fail in the fast test before the broker assertion. Restore the field and confirm the job and report artifact succeed.

Troubleshooting

The test times out although the producer send succeeded -> Inspect consumer.failures(), group ID, topic spelling, and auto.offset.reset. Producer acknowledgement only proves storage. Use a new group with earliest for isolated tests.

Testcontainers cannot connect to Docker -> Run docker version from the same user and CI context. Check the supported runtime configuration and never hide the failure by pointing tests at a shared Kafka cluster.

The consumer misses the first record -> A subscription becomes active during poll. Using earliest with a unique group makes an already-published test record readable. For tighter orchestration, wait until partition assignment is observed before publishing.

Malformed events repeat forever -> Your production error policy may avoid committing the failed offset. Define bounded retries and a durable dead-letter route, then test both. Follow the dead-letter queue and retry testing tutorial for that workflow.

Tests pass locally but fail intermittently in CI -> Replace fixed sleeps with bounded condition polling, give the job enough CPU, reuse one broker per test class, and keep unique consumer groups. Include observed failures and container logs in the assertion output.

Unknown fields break deserialization -> Configure the consumer to tolerate additive fields, or use a schema evolution rule that permits them. Do not weaken required-field or semantic validation merely to accept every payload.

Where To Go Next

Return to the complete event-driven API testing strategy to place this consumer suite within producer, schema, observability, replay, and resilience testing.

Next, automate Avro schema compatibility checks in CI if your events use Schema Registry. Then test dead-letter queue retries and poison events so invalid records cannot stall a partition or disappear silently. If your architecture mixes brokers, the RabbitMQ event mocking tutorial shows how the same boundary-first thinking applies to exchanges and queues.

Treat the contract fixture as executable documentation. Ask producer teams to run the fast consumer contract cases before publishing, but keep broker-backed tests owned by the consumer so its real configuration remains under test.

Interview Questions and Answers

Q: What is a Kafka consumer contract test?

It verifies the record shape and semantics a consumer promises to accept, including topic, key, headers, payload, and resulting behavior. A strong test also covers incompatible and duplicate records.

Q: Why use a real broker container instead of mocking KafkaConsumer?

A container exposes subscription, serialization, polling, partition, header, and offset configuration mistakes. A mock is useful for narrow logic tests but cannot establish transport integration.

Q: Should unknown JSON fields fail a consumer test?

Usually no. Tolerating additive fields lets producers evolve without breaking older consumers, while required fields and semantic invariants should remain strict.

Q: What should the final assertion verify?

Assert the observable business outcome, such as a persisted order or emitted downstream event. Delivery and deserialization are intermediate facts, not the user's outcome.

Q: How do you test at-least-once delivery?

Publish the same event ID more than once and prove only one durable business effect occurs. Production idempotency should be atomic with the write, commonly through a unique constraint or ledger.

Q: Where does schema compatibility testing fit?

It is a fast CI gate that detects structural evolution problems. Broker-backed consumer tests add keys, headers, configuration, and behavioral verification.

Best Practices

• Write the boundary table before writing code. Include topic, key, headers, payload, and failure policy.
• Keep serialization tests fast and deterministic. Run broker-backed tests as a separate, visible layer.
• Assert state changes or emitted events, not private listener calls.
• Use unique groups and isolated topics or records to prevent cross-test contamination.
• Test additive evolution, required-field removal, type changes, semantic invalidity, and duplicates.
• Preserve topic, partition, offset, key, event ID, and failure category in diagnostics.
• Pin broker images and client versions, then exercise upgrades through the suite.
• Never treat a successful producer acknowledgement as proof of successful consumption.

Conclusion

To test Kafka consumer contracts step by step, start with an explicit boundary, prove mapping behavior quickly, then exercise the same event through a real broker and assert a business result. Negative and duplicate cases turn a happy-path integration check into a meaningful compatibility safety net.

Run the suite in CI and pair it with schema compatibility and dead-letter testing. That combination gives producers freedom to evolve while giving consumers evidence that failures are visible, contained, and recoverable.

Interview Questions and Answers

Frequently Asked Questions

Related Guides

• Azure DevOps test pipelines: Step by Step (2026)
• CircleCI for test automation: Step by Step (2026)
• Flaky test quarantine in CI: Step by Step (2026)
• GitLab CI for test automation: Step by Step (2026)
• Grafana dashboards for test metrics: Step by Step (2026)
• Parallel test sharding in CI: Step by Step (2026)