kafka-consumer-tests

kafka-consumer-tests

Overview

Apache Kafka is a distributed event-streaming platform. Unlike SQS (visibility-timeout), RabbitMQ (ack/nack), or BullMQ (job-state machine), Kafka uses consumer-owned committed offsets as its delivery primitive - every test must control where the read position is. This skill covers that model: offset management, consumer-group rebalance, at-least-once vs exactly-once (EOS / transactions), idempotent producer, and dead-letter routing across KafkaJS (Node.js), kafka-go (Go), and Spring Kafka (Java).

Step 1 - Spin up Kafka with Testcontainers

Per testcontainers.com/modules/kafka:

Node.js / TypeScript

npm install @testcontainers/kafka --save-dev

import { KafkaContainer } from '@testcontainers/kafka';

let container: Awaited<ReturnType<KafkaContainer['start']>>;

beforeAll(async () => {
  container = await new KafkaContainer('confluentinc/cp-kafka:7.2.2').start();
});

afterAll(async () => {
  await container.stop();
});

function brokers(): string[] {
  return [`${container.getHost()}:${container.getMappedPort(9093)}`];
}

Per node.testcontainers.org/modules/kafka, the mapped port is 9093; retrieve the host with container.getHost().

Java (Maven)

Per java.testcontainers.org/modules/kafka:

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers-kafka</artifactId>
    <version>2.0.5</version>
    <scope>test</scope>
</dependency>

KafkaContainer kafka = new KafkaContainer("apache/kafka-native:3.8.0");
kafka.start();
String bootstrapServers = kafka.getBootstrapServers();

Go

Per testcontainers.com/modules/kafka:

import "github.com/testcontainers/testcontainers-go/modules/kafka"

kafkaContainer, err := kafka.Run(
    context.Background(),
    "confluentinc/confluent-local:7.5.0",
)

Python

from testcontainers.kafka import KafkaContainer

with KafkaContainer() as kafka:
    bootstrap_server = kafka.get_bootstrap_server()

Step 2 - Basic producer test (KafkaJS)

Per kafka.js.org/docs/getting-started:

import { Kafka } from 'kafkajs';

it('produces a message to the topic', async () => {
  const kafka = new Kafka({ brokers: brokers() });
  const producer = kafka.producer();
  await producer.connect();

  const result = await producer.send({
    topic: 'orders',
    messages: [{ key: 'order-1', value: JSON.stringify({ amount: 42 }) }],
  });

  expect(result[0].errorCode).toBe(0);
  await producer.disconnect();
});

producer.send() returns partition metadata per kjs-gs; errorCode: 0 confirms the broker accepted the record.

Step 3 - Basic consumer test: offset management

Per kafka.js.org/docs/consuming:

Two commit strategies:

Strategy	KafkaJS config	Risk
Auto-commit	autoCommit: true (default)	offset committed before processing finishes - can skip on crash
Manual commit	autoCommit: false + commitOffsetsIfNecessary()	offset committed only after processing - at-least-once

it('consumes messages and commits offsets manually', async () => {
  const kafka = new Kafka({ brokers: brokers() });
  const received: string[] = [];

  const consumer = kafka.consumer({ groupId: 'test-group-1' });
  await consumer.connect();
  await consumer.subscribe({ topic: 'orders', fromBeginning: true });

  await consumer.run({
    autoCommit: false,
    eachBatch: async ({ batch, commitOffsetsIfNecessary, resolveOffset }) => {
      for (const msg of batch.messages) {
        received.push(msg.value!.toString());
        resolveOffset(msg.offset);          // mark processed
      }
      await commitOffsetsIfNecessary();     // commit atomically
    },
  });

  // wait for consumption then assert
  await new Promise(r => setTimeout(r, 1000));
  expect(received).toContain(JSON.stringify({ amount: 42 }));
  await consumer.disconnect();
});

fromBeginning: true sets auto.offset.reset to earliest per kjs-consuming: "starts from earliest [offset] when no committed offset exists."

Step 4 - Consumer-group rebalance

Per kafka.js.org/docs/consuming, the three timing parameters that govern rebalance are:

Parameter	Default	Purpose
sessionTimeout	30 000 ms	failure-detection window; broker removes member if no heartbeat within this window
rebalanceTimeout	60 000 ms	time allowed for members to rejoin after a rebalance is triggered
heartbeatInterval	3 000 ms	frequency of liveness pings to the broker

A rebalance test starts two consumers in the same group, sends messages, stops one consumer, then asserts the survivor processes all remaining messages:

it('surviving consumer processes all messages after rebalance', async () => {
  const kafka = new Kafka({ brokers: brokers() });
  const groupId = 'rebalance-test-group';

  const c1 = kafka.consumer({ groupId, sessionTimeout: 6000, heartbeatInterval: 1000 });
  const c2 = kafka.consumer({ groupId, sessionTimeout: 6000, heartbeatInterval: 1000 });
  const received: string[] = [];

  for (const c of [c1, c2]) {
    await c.connect();
    await c.subscribe({ topic: 'events', fromBeginning: true });
  }
  c2.run({ eachMessage: async ({ message }) => { received.push(message.value!.toString()); } });
  c1.run({ eachMessage: async ({ message }) => { received.push(message.value!.toString()); } });

  await c1.disconnect();   // trigger rebalance: c2 inherits all partitions

  // send more messages after rebalance
  const producer = kafka.producer();
  await producer.connect();
  await producer.send({ topic: 'events', messages: [{ value: 'post-rebalance' }] });
  await producer.disconnect();

  await new Promise(r => setTimeout(r, 3000));
  expect(received).toContain('post-rebalance');
  await c2.disconnect();
});

Lowering sessionTimeout to 6 000 ms speeds up the rebalance in tests; production values are higher to avoid spurious rebalances.

Step 5 - At-least-once vs exactly-once (EOS / transactions)

At-least-once delivery is the default when autoCommit: false and the processor retries on failure - a crash before commitOffsetsIfNecessary() causes the broker to re-deliver the unacknowledged messages.

Exactly-once semantics (EOS) requires the transactional producer. Per kafka.js.org/docs/transactions:

const producer = kafka.producer({
  transactionalId: 'my-transactional-producer',
  maxInFlightRequests: 1,       // required for EOS
  idempotent: true,             // required for EOS
});
await producer.connect();

const tx = await producer.transaction();
try {
  await tx.send({ topic: 'output', messages: [{ value: 'result' }] });
  // atomically commit the consumer offset with the produced message
  await tx.sendOffsets({ consumerGroupId: 'test-group', topics: [...] });
  await tx.commit();
} catch (err) {
  await tx.abort();
  throw err;
}

Per kjs-tx: "The producer must have a max in flight requests of 1" and must use acks: -1 (all replicas). The transactionalId should encode topic and partition (e.g. 'myapp-producer-topic-0') so the broker can fence out duplicate instances.

A consumer must use readUncommitted: false (the default) to exclude in-flight transactional messages per kjs-consuming: "Set readUncommitted: false (default) to exclude uncommitted transactional messages from consumption."

EOS test pattern:

it('exactly-once: aborted transaction produces no visible messages', async () => {
  const producer = kafka.producer({
    transactionalId: 'eos-test',
    maxInFlightRequests: 1,
    idempotent: true,
  });
  await producer.connect();

  const tx = await producer.transaction();
  await tx.send({ topic: 'eos-topic', messages: [{ value: 'ghost' }] });
  await tx.abort();   // consumer should never see 'ghost'

  const consumer = kafka.consumer({ groupId: 'eos-verify' });
  await consumer.connect();
  await consumer.subscribe({ topic: 'eos-topic', fromBeginning: true });
  const received: string[] = [];
  consumer.run({ eachMessage: async ({ message }) => { received.push(message.value!.toString()); } });

  await new Promise(r => setTimeout(r, 1500));
  expect(received).not.toContain('ghost');

  await consumer.disconnect();
  await producer.disconnect();
});

Step 6 - Idempotent producer (standalone)

Per kafka.js.org/docs/transactions, idempotent: true can be set without a transactionalId. The broker deduplicates retried produce requests within the same producer session and sequence number:

const producer = kafka.producer({ idempotent: true });

This is sufficient to prevent duplicate records from network-level retries. It does not deduplicate across producer restarts - use the transactional API (Step 5) or application-level idempotency (see idempotency-test-author) for cross-restart deduplication.

Step 7 - Dead-letter topic (DLT)

KafkaJS provides no built-in DLT routing per the consuming documentation - the application must catch errors in eachMessage / eachBatch and produce to a <topic>.DLT topic manually:

const consumer = kafka.consumer({ groupId: 'dlt-group' });
const dlqProducer = kafka.producer();
await consumer.connect();
await dlqProducer.connect();
await consumer.subscribe({ topic: 'orders', fromBeginning: true });

consumer.run({
  eachMessage: async ({ topic, message }) => {
    try {
      await processOrder(JSON.parse(message.value!.toString()));
    } catch (err) {
      // route poison message to dead-letter topic
      await dlqProducer.send({
        topic: `${topic}.DLT`,
        messages: [{
          key: message.key,
          value: message.value,
          headers: { 'x-error': String(err) },
        }],
      });
    }
  },
});

Test the DLT route:

it('routes unprocessable message to DLT', async () => {
  // arrange: produce a poison message
  await producer.send({ topic: 'orders', messages: [{ value: 'NOT-JSON' }] });

  await new Promise(r => setTimeout(r, 1500));

  // assert: DLT received the record
  const dltConsumer = kafka.consumer({ groupId: 'dlt-verifier' });
  await dltConsumer.connect();
  await dltConsumer.subscribe({ topic: 'orders.DLT', fromBeginning: true });
  const dltMessages: string[] = [];
  dltConsumer.run({ eachMessage: async ({ message }) => {
    dltMessages.push(message.value!.toString());
  }});
  await new Promise(r => setTimeout(r, 1500));
  expect(dltMessages).toContain('NOT-JSON');
  await dltConsumer.disconnect();
});

Spring Kafka provides @RetryableTopic for declarative DLT routing per docs.spring.io/spring-kafka/reference/testing.html.

Step 8 - Other languages

Java (Spring Kafka + EmbeddedKafka)

Per sk-test, since Kafka 4.0 only EmbeddedKafkaKraftBroker is available (KRaft / no ZooKeeper):

@SpringJUnitConfig
@EmbeddedKafka(partitions = 1, topics = {"orders"},
    bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class OrderConsumerTest {
    @Autowired EmbeddedKafkaBroker embeddedKafka;

    @Test
    void testTemplate() throws Exception {
        Map<String, Object> cp = KafkaTestUtils.consumerProps("tg", "false", embeddedKafka);
        cp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // default earliest since 2.5
        Consumer<Integer, String> consumer =
            new DefaultKafkaConsumerFactory<Integer, String>(cp).createConsumer();
        embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "orders");

        KafkaTemplate<Integer, String> template = new KafkaTemplate<>(
            new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka)), true);
        template.send("orders", "test-order");

        ConsumerRecord<Integer, String> received =
            KafkaTestUtils.getSingleRecord(consumer, "orders");
        assertThat(received).has(value("test-order"));
    }
}

Go (kafka-go + Testcontainers)

Per pkg.go.dev/github.com/segmentio/kafka-go:

func TestRoundTrip(t *testing.T) {
    ctx := context.Background()
    kc, _ := kafka.Run(ctx, "confluentinc/confluent-local:7.5.0")
    defer kc.Terminate(ctx)
    addr := kc.MustConnectionString(ctx)

    w := &kafkago.Writer{Addr: kafkago.TCP(addr), Topic: "events",
        AllowAutoTopicCreation: true}
    w.WriteMessages(ctx, kafkago.Message{Value: []byte("hello")})
    w.Close()

    r := kafkago.NewReader(kafkago.ReaderConfig{
        Brokers: []string{addr}, GroupID: "tg", Topic: "events"})
    m, _ := r.FetchMessage(ctx)          // FetchMessage = manual commit (at-least-once)
    require.Equal(t, "hello", string(m.Value))
    r.CommitMessages(ctx, m)
    r.Close()
}

Per kgo: FetchMessage + CommitMessages is the explicit commit path; ReadMessage auto-commits and is at-most-once on crash.

Step 9 - CI integration

Testcontainers pulls the broker image at runtime; no services: block is needed. Example for Node.js (GitHub Actions):

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with: { node-version: '20' }
      - run: npm ci
      - run: npx jest --testPathPattern=kafka
        env:
          TESTCONTAINERS_RYUK_DISABLED: 'true'

For Java, replace the Node steps with actions/setup-java and mvn test. For Go, replace with actions/setup-go and go test.

Anti-patterns

Anti-pattern	Why it fails	Fix
Use ReadMessage (kafka-go) for at-least-once tests	Commits before caller processes; at-most-once on crash	Use FetchMessage + CommitMessages
Share groupId across parallel tests	Rebalance storms; flaky message allocation	Use unique groupId per test (e.g. append uuid)
Skip await producer.disconnect() / r.Close()	Goroutine / connection leaks; CI hangs	Close in afterAll / defer
Assert message count with autoCommit: true	Offset committed before processing; race on crash path	Use autoCommit: false + manual commit for assertion tests
Test EOS with readUncommitted: true	Consumer sees aborted messages; false-passing test	Keep readUncommitted: false (default)
Use sessionTimeout > rebalanceTimeout	Rebalance never completes	Ensure sessionTimeout < rebalanceTimeout

Limitations

• Testcontainers Kafka image pull adds 30-60 s on a cold CI runner; cache the Docker layer to mitigate.
• kafka-go does not expose a transactional producer API per kgo; EOS tests in Go require the confluent-kafka-go library or a higher-level framework.
• Spring Kafka's EmbeddedKafkaKraftBroker (Kafka 4.0+) does not support the old ZooKeeper-based configuration; update spring-kafka-test to 4.x if migrating from older Spring Boot.
• Consumer-group rebalance test timing is environment-dependent; lower sessionTimeout in tests (e.g. 6 000 ms) rather than relying on wall-clock sleep durations.

References

• testcontainers.com/modules/kafka - module coordinates for Java, Node.js, Go, Python, .NET; container lifecycle
• node.testcontainers.org/modules/kafka - Node.js KafkaContainer API, mapped port, SSL/SASL options
• java.testcontainers.org/modules/kafka - Java KafkaContainer / ConfluentKafkaContainer, getBootstrapServers()
• kafka.js.org/docs/getting-started - KafkaJS install, producer send() API
• kafka.js.org/docs/consuming - eachMessage / eachBatch, fromBeginning, auto-commit vs manual commit, sessionTimeout, rebalanceTimeout, readUncommitted
• kafka.js.org/docs/transactions - transactional producer, idempotent: true, maxInFlightRequests: 1, sendOffsets(), commit() / abort()
• pkg.go.dev/github.com/segmentio/kafka-go - Reader, Writer, FetchMessage, CommitMessages, GroupID, CommitInterval
• docs.spring.io/spring-kafka/reference/testing.html - @EmbeddedKafka, KafkaTestUtils, MockConsumer, MockProducer, Spring Kafka 4.x / KRaft broker
• sqs-patterns, rabbitmq-patterns, bullmq-tests - sibling broker skills
• idempotency-test-author - application-level idempotency patterns (complements Step 6)