Testing Kafka applications: patterns, pitfalls and fixes
Most teams only discover their Kafka wiring works when it hits production. By then, a cascading rebalance, a malformed event, or a subtle ordering bug has already cost them. The gap between "my handler processes one message correctly" and "my entire pipeline survives the chaos of the real world" is where testing Kafka applications: patterns, pitfalls and fixes becomes indispensable. We'll walk through how to build confidence without needing a shared cluster, and show you exactly where most teams go wrong.
Unit test handlers — Keep business logic separate from Kafka plumbing
Your first line of defense is to isolate business logic from Kafka's transport layer. A handler should accept a domain object and return a result. The fact that it arrived via a Kafka message is an implementation detail.
Consider an order event handler. Instead of writing tests that spin up a broker or mock a KafkaConsumer, write tests that call the handler directly with a plain object:
// ✓ Good: test business logic in isolation
describe('OrderProcessor', () => {
it('should mark order as confirmed when stock available', () => {
const handler = new OrderProcessor(mockInventoryService);
const event = {
orderId: '12345',
sku: 'WIDGET-A',
quantity: 5,
};
const result = handler.process(event);
expect(result).toEqual({
status: 'confirmed',
confirmationId: expect.any(String),
});
});
it('should reject order when stock depleted', () => {
mockInventoryService.getStock.mockResolvedValue(2);
const handler = new OrderProcessor(mockInventoryService);
const event = {
orderId: '12345',
sku: 'WIDGET-A',
quantity: 5,
};
expect(handler.process(event)).rejects.toThrow('Insufficient stock');
});
});
Keep Kafka-specific concerns—serialization, offset management, error callbacks—in a thin adapter layer that wraps your handlers. Test the adapter separately with minimal Kafka surface area. This separation means your business tests run in milliseconds and never depend on broker availability.
A common mistake here is embedding Kafka metadata (partition, offset, timestamp) directly into your handler logic. If you need that information, pass it as a separate parameter and make it optional for unit tests.
Contract tests — Validate events against registered schemas
Kafka is a protocol for asynchronous systems. Your consumer has no guarantee that the producer sent the shape of data it expects. Schema drift breaks silently; a missing field gets parsed as null, and suddenly your order processor is confirming orders for quantity zero.
Contract tests enforce a schema contract between producer and consumer without requiring both to run together. You register schemas in a schema registry—Confluent Schema Registry, AWS Glue, or Protobuf definitions—and both sides validate against it.
// Schema Registry defines this shape
const OrderCreatedSchema = {
type: 'object',
properties: {
orderId: { type: 'string' },
sku: { type: 'string' },
quantity: { type: 'integer', minimum: 1 },
customerId: { type: 'string' },
timestamp: { type: 'number' },
},
required: ['orderId', 'sku', 'quantity', 'customerId', 'timestamp'],
};
// Consumer contract test: validate inbound messages conform to schema
describe('OrderCreatedEvent Contract', () => {
it('should accept valid order created event', async () => {
const validator = new SchemaValidator(OrderCreatedSchema);
const event = {
orderId: 'ORD-001',
sku: 'WIDGET-A',
quantity: 3,
customerId: 'CUST-456',
timestamp: Date.now(),
};
expect(() => validator.validate(event)).not.toThrow();
});
it('should reject event missing required quantity field', async () => {
const validator = new SchemaValidator(OrderCreatedSchema);
const event = {
orderId: 'ORD-001',
sku: 'WIDGET-A',
// quantity missing
customerId: 'CUST-456',
timestamp: Date.now(),
};
expect(() => validator.validate(event)).toThrow('quantity');
});
it('should reject quantity of zero', async () => {
const validator = new SchemaValidator(OrderCreatedSchema);
const event = {
orderId: 'ORD-001',
sku: 'WIDGET-A',
quantity: 0,
customerId: 'CUST-456',
timestamp: Date.now(),
};
expect(() => validator.validate(event)).toThrow('minimum');
});
});
The producer team runs an identical contract test to ensure they never publish data that violates the schema. This catches mismatches before they reach staging or production. Use a shared schema repository so both teams depend on the same source of truth.
Integration tests — Testcontainers starts a real broker for end-to-end flows
Unit and contract tests catch most logic bugs. Integration tests verify that your entire pipeline—producer, broker, consumer, state changes—works as a cohesive system. This is where you test ordering guarantees, exactly-once semantics, and interactions with downstream services.
Testcontainers spins up a real Kafka broker (and PostgreSQL, Redis, etc.) in a container for the duration of your test. No mocking, no in-memory substitutes. You get a genuine broker with real rebalancing, real offset tracking, and real timeouts.
import { GenericContainer } from 'testcontainers';
describe('Order Processing Pipeline - Integration', () => {
let kafkaContainer;
let kafka: Kafka;
let producer: Producer;
let consumer: Consumer;
beforeAll(async () => {
// Start a real Kafka broker
kafkaContainer = await new GenericContainer('confluentinc/cp-kafka:7.5.0')
.withExposedPorts(9092)
.withEnvironment({
KAFKA_ZOOKEEPER_CONNECT: 'localhost:2181',
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://localhost:9092',
})
.start();
const brokerUrl = `${kafkaContainer.getHost()}:${kafkaContainer.getMappedPort(9092)}`;
kafka = new Kafka({ brokers: [brokerUrl] });
producer = kafka.producer();
consumer = kafka.consumer({ groupId: 'test-group' });
await producer.connect();
await consumer.connect();
});
afterAll(async () => {
await producer.disconnect();
await consumer.disconnect();
await kafkaContainer.stop();
});
it('should process order from creation to confirmation', async () => {
const topic = 'orders';
await admin.createTopics({
topics: [{ name: topic, numPartitions: 3, replicationFactor: 1 }],
});
// Produce an order event
await producer.send({
topic,
messages: [
{
key: 'ORD-001',
value: JSON.stringify({
orderId: 'ORD-001',
sku: 'WIDGET-A',
quantity: 5,
customerId: 'CUST-456',
timestamp: Date.now(),
}),
},
],
});
// Consume and process
let processedOrder;
await consumer.subscribe({ topic, fromBeginning: true });
await consumer.run({
eachMessage: async ({ message }) => {
const event = JSON.parse(message.value.toString());
processedOrder = await orderProcessor.process(event);
},
});
// Verify state persisted
expect(processedOrder.status).toBe('confirmed');
const persisted = await db.orders.findOne({ id: 'ORD-001' });
expect(persisted.status).toBe('confirmed');
});
});
Integration tests are slower (seconds, not milliseconds) and more brittle, so keep them focused. Test the happy path once, then move failure scenarios to the next tier.
Test failure paths — Duplicates, poison messages and rebalances
Production Kafka is messy. Networks partition, brokers restart, and your consumer crashes mid-processing. The code that handles these scenarios is often untested, which means it's often broken.
Duplicate messages
Kafka offers at-most-once and at-least-once delivery. Most teams choose at-least-once, which means the same message can arrive twice. Your handler must be idempotent.
describe('Idempotency - Duplicate Messages', () => {
it('should not double-charge customer for duplicate order message', async () => {
const handler = new OrderProcessor(mockInventoryService, db);
const event = {
orderId: 'ORD-001',
amount: 99.99,
customerId: 'CUST-456',
};
// First delivery
const result1 = await handler.process(event);
expect(result1.status).toBe('confirmed');
// Duplicate arrives (simulating redelivery after crash)
const result2 = await handler.process(event);
expect(result2.status).toBe('confirmed');
// Only one charge
const charges = await db.charges.find({ orderId: 'ORD-001' });
expect(charges).toHaveLength(1);
});
});
Store a deduplication key (usually the message key or a unique event ID) in your state and skip processing if you've seen it before.
Poison messages
A poison message is one your consumer cannot process—malformed JSON, missing required fields, or a downstream service permanently rejecting it. A naive consumer spins in an infinite retry loop. A well-designed one moves the message to a dead-letter topic and continues.
describe('Dead-Letter Topic - Poison Messages', () => {
it('should route unparseable message to dead-letter topic', async () => {
const handler = new OrderProcessor();
const deadLetterTopic = 'orders-dead-letter';
const malformedMessage = Buffer.from('{ invalid json');
try {
JSON.parse(malformedMessage.toString());
} catch (error) {
// Send to dead-letter with metadata for debugging
await producer.send({
topic: deadLetterTopic,
messages: [
{
value: malformedMessage,
headers: {
'original-error': Buffer.from(error.message),
'failed-at': Buffer.from(new Date().toISOString()),
},
},
],
});
}
// Verify consumer continues processing
const liveMessages = [/* valid orders */];
for (const msg of liveMessages) {
expect(() => handler.process(msg)).not.toThrow();
}
});
});
Rebalances
When a consumer joins or leaves a group, Kafka rebalances partitions. During a rebalance, your consumer stops processing. If you've committed offsets for partially processed batches, you'll replay messages on recovery.
describe('Rebalance Handling', () => {
it('should commit offsets only after processing completes', async () => {
const handler = new OrderProcessor(db);
let processedCount = 0;
consumer.on('consumer.crash', (error) => {
// If we crash here, the last uncommitted offset is replayed
console.error('Consumer crashed during rebalance', error);
});
await consumer.run({
eachMessage: async ({ topic, partition, message }) => {
try {
const event = JSON.parse(message.value.toString());
await handler.process(event);
processedCount++;
// Only commit after success
await consumer.commitOffsets([
{
topic,
partition,
offset: (BigInt(message.offset) + BigInt(1)).toString(),
},
]);
} catch (error) {
console.error('Processing failed', error);
// Do not commit; message will be replayed
throw error;
}
},
});
// Simulate consumer crash before rebalance completes
// Expect messages to be replayed but not duplicated in DB
});
});
Test this by simulating consumer crashes, forcing rebalances, and verifying that messages are neither lost nor processed twice.
Common mistakes
- Testing only the happy path. A handler that works for one message may crash on the hundredth, or fail when the database is slow. Test error branches explicitly.
- Mocking the Kafka client instead of the business dependencies. Mocking KafkaProducer or KafkaConsumer makes your tests pass locally but false when behavior drifts from the real client. Mock your database, HTTP clients, and third-party services instead.
- Ignoring schema evolution. Your schema will change. Test that old messages can still be parsed by new code, and new messages are rejected by old code gracefully.
- Skipping offset management tests. Offset commits are often the hidden failure point. Test that offsets are committed atomically with state changes, not before.
- Running all tests against a shared staging broker. Flaky network, test interference, and blocked ports turn tests into a bottleneck. Use Testcontainers for isolation.
Key takeaways
- Separate business logic from Kafka plumbing so unit tests run fast and don't depend on broker availability.
- Use contract tests with a schema registry to catch data shape mismatches before they reach production.
- Spin up a real Kafka broker with Testcontainers for integration tests; this catches ordering, rebalancing, and exactly-once issues that mocks miss.
- Explicitly test failure paths: duplicates, poison messages, and rebalances. These are where most production outages hide.
- Commit offsets only after successful processing, and test that messages are replayed correctly when consumers crash.
- Mock downstream services and databases, not Kafka itself. Your tests should be indistinguishable from production behavior.
Comments
Loading comments…