Testing Kafka Consumers With an In-Memory Broker

Kafka consumers are usually tested in one of two unsatisfying ways: against a real cluster that takes thirty seconds to start and behaves differently every run, or not at all beyond the handler function. The first makes the suite slow and flaky; the second leaves the consumer’s own responsibilities — decoding, per-partition ordering, committing offsets only after success, redelivering after failure — completely unverified. This guide builds a small in-memory broker that models the parts of Kafka a consumer depends on, uses it to test those responsibilities deterministically, and keeps a handful of tests against a real broker for the client library itself. It targets KafkaJS 2.x with Vitest, and sits under event-driven and queue mocking.

Root Cause Analysis

A Kafka consumer has more responsibility than it appears to. Beyond calling a handler for each message, it decides when a message counts as done — by committing its offset — and that decision determines what happens after a crash. Commit before the handler finishes and a crash loses the message; commit after and a crash redelivers it. Getting this wrong is invisible in normal operation and catastrophic during an incident, and it is precisely the behaviour a handler-only test cannot see.

Ordering is the second responsibility. Kafka guarantees order within a partition, not across the topic, and a consumer that processes a partition’s messages concurrently breaks the guarantee its producers were relying on. A test with a single message cannot detect that; a test with several messages for the same key, processed with realistic concurrency, can.

A real cluster exercises both, but it does so non-deterministically. Partition assignment, rebalance timing and batch boundaries vary run to run, so a test that relies on them is flaky by construction. An in-memory model that captures only partitions, offsets and commits — and lets the test decide when things happen — turns those behaviours into ordinary, reproducible assertions.

Reproducible Setup

Keep the consumer’s logic in a function that depends on a small interface rather than on the KafkaJS client directly. The real adapter implements the interface with KafkaJS; tests implement it in memory.

// src/messaging/consumer-port.ts
export type Record = { topic: string; partition: number; offset: string; key: string; value: string };

export interface ConsumerPort {
  run(eachMessage: (r: Record) => Promise<void>): Promise<void>;
  commit(r: Record): Promise<void>;
}

// src/orders/consume-orders.ts — the consumer under test
import type { ConsumerPort } from '../messaging/consumer-port';
import { orderEventSchema } from './events';
import { onOrderEvent } from './handlers';

export function consumeOrders(port: ConsumerPort, deps: Deps) {
  return port.run(async (record) => {
    const parsed = orderEventSchema.safeParse(JSON.parse(record.value));
    if (!parsed.success) {
      await deps.deadLetter.put(record, 'schema_invalid');
      await port.commit(record);                       // do not block the partition on a poison message
      return;
    }
    await onOrderEvent(parsed.data, deps);             // throws on failure → no commit → redelivery
    await port.commit(record);
  });
}

Implementation

Step 1 — Model partitions and offsets in memory. The broker keeps an append-only log per partition and a committed offset per partition; that is all a consumer’s correctness depends on.

// test/fakes/memory-kafka.ts
import type { ConsumerPort, Record } from '../../src/messaging/consumer-port';

export function memoryKafka(partitions = 3) {
  const logs: Record[][] = Array.from({ length: partitions }, () => []);
  const committed = new Array<number>(partitions).fill(-1);
  const partitionFor = (key: string) => [...key].reduce((h, c) => (h * 31 + c.charCodeAt(0)) >>> 0, 7) % partitions;

  return {
    produce(topic: string, key: string, value: unknown) {
      const p = partitionFor(key);
      logs[p].push({ topic, partition: p, offset: String(logs[p].length), key, value: JSON.stringify(value) });
    },
    committed: () => [...committed],
    port(): ConsumerPort & { drain(): Promise<void> } {
      let handler: (r: Record) => Promise<void>;
      return {
        async run(h) { handler = h; },
        async commit(r) { committed[r.partition] = Number(r.offset); },
        async drain() {
          for (let p = 0; p < partitions; p++) {
            for (const r of logs[p].slice(committed[p] + 1)) {
              try { await handler(r); } catch { break; }   // stop this partition, as Kafka would
            }
          }
        },
      };
    },
  };
}

Step 2 — Assert that offsets advance only after success. A failing handler must leave the offset where it was, so the next poll redelivers the same message.

// src/orders/consume-orders.test.ts
import { test, expect } from 'vitest';
import { memoryKafka } from '../../test/fakes/memory-kafka';
import { consumeOrders } from './consume-orders';
import { memoryDeps } from '../../test/fakes/order-deps';

test('does not commit when the handler fails, so the message is redelivered', async () => {
  const kafka = memoryKafka(1);
  const deps = memoryDeps();
  deps.invoices.failNext();

  kafka.produce('orders', 'o1', { type: 'OrderPlaced', eventId: 'e1', orderId: 'o1', totalPence: 1000 });
  const port = kafka.port();
  await consumeOrders(port, deps);

  await port.drain();
  expect(kafka.committed()).toEqual([-1]);           // nothing committed after the failure

  await port.drain();                                 // redelivery
  expect(kafka.committed()).toEqual([0]);
  expect(deps.invoices.all()).toHaveLength(1);
});

Step 3 — Assert per-key ordering. Messages for the same key land on the same partition and must be handled in the order produced, which is the guarantee producers rely on.

test('processes events for one order in the order they were produced', async () => {
  const kafka = memoryKafka(3);
  const deps = memoryDeps();
  for (const type of ['OrderPlaced', 'OrderPaid', 'OrderShipped'] as const) {
    kafka.produce('orders', 'o9', { type, eventId: `${type}-o9`, orderId: 'o9', totalPence: 500 });
  }
  const port = kafka.port();
  await consumeOrders(port, deps);
  await port.drain();

  expect(deps.history('o9')).toEqual(['OrderPlaced', 'OrderPaid', 'OrderShipped']);
});

Step 4 — Assert that a poison message does not block its partition. A payload that can never be processed goes to the dead-letter store and is committed, so later messages on the same partition still flow.

test('dead-letters an invalid payload and carries on with the partition', async () => {
  const kafka = memoryKafka(1);
  const deps = memoryDeps();
  kafka.produce('orders', 'o1', { nonsense: true });
  kafka.produce('orders', 'o1', { type: 'OrderPlaced', eventId: 'e2', orderId: 'o1', totalPence: 100 });

  const port = kafka.port();
  await consumeOrders(port, deps);
  await port.drain();

  expect(deps.deadLetter.all()).toHaveLength(1);
  expect(deps.invoices.all()).toHaveLength(1);
  expect(kafka.committed()).toEqual([1]);
});

Step 5 — Keep a few tests against a real broker. The KafkaJS adapter — connection, group membership, the mapping to the port — needs a real cluster to verify, but only a handful of tests.

// test/integration/kafka-adapter.test.ts
import { KafkaContainer } from '@testcontainers/kafka';

let container: StartedKafkaContainer;
beforeAll(async () => { container = await new KafkaContainer('confluentinc/cp-kafka:7.6.1').start(); }, 120_000);
afterAll(() => container.stop());

test('the adapter commits after the handler resolves', async () => {
  // produce one message, run the real adapter, and read the group's committed offset
});

Step 6 — Namespace topics and consumer groups per run. A shared cluster in CI will otherwise deliver one job’s messages to another job’s consumer.

const run = process.env.GITHUB_RUN_ID ?? 'local';
const topic = `orders-${run}-${process.pid}`;
const groupId = `orders-consumer-${run}-${process.pid}`;

Verification

Run the in-memory suite and confirm it covers each consumer responsibility, not only the happy path.

npx vitest run src/orders/consume-orders.test.ts --reporter=verbose
# ✓ does not commit when the handler fails, so the message is redelivered
# ✓ processes events for one order in the order they were produced
# ✓ dead-letters an invalid payload and carries on with the partition

Then prove the commit test is meaningful by moving the commit before the handler call in the consumer. The redelivery test must fail; if it still passes, the fake is not modelling commits faithfully.

Troubleshooting

Symptom: a failed message is never retried in the in-memory tests. Diagnosis: the fake commits unconditionally, or the consumer swallows the handler’s error. Fix: let handler errors propagate to the port, and have the fake stop draining a partition at the first failure, as Kafka would.

Symptom: ordering tests pass but production processes a key out of order. Diagnosis: the real consumer uses a batch or concurrency setting that processes several messages from one partition at once. Fix: configure per-partition concurrency to one for order-sensitive topics, and assert that setting in the adapter’s own test.

Symptom: the containerised tests time out waiting for messages. Diagnosis: the consumer joined the group after the producer wrote, and it starts from the latest offset. Fix: set fromBeginning: true for test consumers, or subscribe before producing.

Symptom: rebalances cause duplicate processing in the real-broker tests. Diagnosis: expected behaviour — a rebalance can redeliver uncommitted messages. Fix: this is what the idempotency tests exist for; assert on side-effect counts rather than on handler invocation counts in these tests.

FAQ

Related

• Back to Event-Driven & Queue Mocking
• Mocking SQS and SNS clients in Vitest — the same principles for AWS queues.
• Testing background jobs queued with BullMQ — job queues rather than event logs.
• Eliminating test order dependence — why the fake makes delivery order explicit.