Poison pills
One malformed event can stop a consumer forever. You process a message, the deserializer throws an exception, the consumer crashes or gets stuck in a retry loop, and your entire pipeline grinds to a halt. These problematic messages are called poison pills—they're toxic to your system because they block progress and can go unnoticed for hours.
In this article, I'll walk you through how to identify poison pills early, isolate them, and keep your Kafka consumers resilient. This is one of those unglamorous but critical patterns you'll thank yourself for implementing when production breaks at 2 AM.
Understanding the poison pill problem
Picture this: your order processing consumer reads events from Kafka. One message arrives with a malformed JSON payload—maybe a schema version mismatch, a truncated field, or corrupted binary data. Your deserializer explodes. The consumer logs the error, rebalances, retries the same message, and crashes again. You've created an infinite loop of failures.
The consumer never commits the offset, so it keeps replaying the same poisoned message. Meanwhile, your real orders pile up downstream, customers don't get their confirmations, and nobody realizes it's because a single bad message blocked the entire queue.
The solution isn't to ignore these messages or crash loudly. It's to catch them explicitly, route them somewhere safe, and keep moving forward.
Catch deserialization errors explicitly
The first step is to wrap your deserialization logic in a try-catch block. Instead of letting exceptions bubble up to Kafka's error handler, you intercept them and decide what to do.
Here's a real example using NestJS with the KafkaJS library:
import { Injectable } from '@nestjs/common';
import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';
@Injectable()
export class OrderConsumer {
private consumer: Consumer;
constructor(private kafka: Kafka) {
this.consumer = this.kafka.consumer({ groupId: 'order-processor' });
}
async handleOrderCreated(payload: EachMessagePayload): Promise {
const { topic, partition, offset, value } = payload;
try {
// Attempt to deserialize the message
const order = JSON.parse(value.toString());
// Validate the schema
if (!order.id || !order.customerId || !order.amount) {
throw new Error('Invalid order schema');
}
// Process the valid order
await this.processOrder(order);
console.log(`Processed order ${order.id} at offset ${offset}`);
} catch (error) {
// Catch both deserialization and validation errors
console.error(`Failed to process message at offset ${offset}:`, error.message);
// Log the raw message for debugging
console.error(`Raw payload: ${value.toString()}`);
// Don't rethrow—let the consumer continue
// We'll handle routing to DLQ in the next section
}
}
async subscribe(): Promise {
await this.consumer.connect();
await this.consumer.subscribe({ topic: 'orders' });
await this.consumer.run({
eachMessage: this.handleOrderCreated.bind(this),
});
}
}
The key insight here is don't rethrow the error. If you throw, Kafka will retry forever. Instead, log it with enough context (the raw payload, the offset, the partition) so you can debug later.
Send bad messages to a Dead Letter Queue
Catching the error silently isn't enough—you need a quarantine zone. A Dead Letter Queue (DLQ) is a separate Kafka topic where you route poisoned messages along with metadata about what went wrong.
Here's how to implement it:
import { Injectable } from '@nestjs/common';
import { Kafka, Consumer, Producer, EachMessagePayload } from 'kafkajs';
interface DeadLetter {
originalTopic: string;
originalPartition: number;
originalOffset: number;
originalTimestamp: string;
rawPayload: string;
error: string;
errorTimestamp: string;
}
@Injectable()
export class OrderConsumerWithDLQ {
private consumer: Consumer;
private dlqProducer: Producer;
constructor(private kafka: Kafka) {
this.consumer = this.kafka.consumer({ groupId: 'order-processor' });
this.dlqProducer = this.kafka.producer();
}
private async sendToDLQ(
topic: string,
partition: number,
offset: number,
timestamp: string,
rawPayload: string,
error: Error
): Promise {
const deadLetter: DeadLetter = {
originalTopic: topic,
originalPartition: partition,
originalOffset: offset,
originalTimestamp: timestamp,
rawPayload,
error: error.message,
errorTimestamp: new Date().toISOString(),
};
try {
await this.dlqProducer.send({
topic: 'orders-dlq',
messages: [
{
key: `${topic}-${partition}-${offset}`,
value: JSON.stringify(deadLetter),
headers: {
'original-topic': topic,
'error-reason': error.message,
},
},
],
});
console.log(`Sent poison pill to DLQ: offset ${offset}`);
} catch (dlqError) {
// If DLQ itself fails, log loudly and alert
console.error('CRITICAL: Failed to send to DLQ:', dlqError);
// Send to your alerting system (PagerDuty, Slack, etc.)
}
}
async handleOrderCreated(payload: EachMessagePayload): Promise {
const { topic, partition, offset, value, timestamp } = payload;
try {
const order = JSON.parse(value.toString());
if (!order.id || !order.customerId || !order.amount) {
throw new Error('Invalid order schema: missing required fields');
}
await this.processOrder(order);
console.log(`Processed order ${order.id}`);
} catch (error) {
console.error(`Poison pill detected at offset ${offset}:`, error.message);
// Route to DLQ
await this.sendToDLQ(
topic,
partition,
offset,
timestamp,
value.toString(),
error
);
}
}
async subscribe(): Promise {
await this.consumer.connect();
await this.dlqProducer.connect();
await this.consumer.subscribe({ topic: 'orders' });
await this.consumer.run({
eachMessage: this.handleOrderCreated.bind(this),
});
}
}
By sending the original payload, offset, partition, and error details to the DLQ, you create a complete audit trail. Later, you can replay specific messages or investigate what went wrong.
Commit past them
Once a message is routed to the DLQ, you must commit the offset. This tells Kafka you've processed that message (even if only by quarantining it), so the consumer won't replay it forever.
In KafkaJS with manual commits, it looks like this:
async handleOrderCreated(payload: EachMessagePayload): Promise {
const { topic, partition, offset, value, timestamp } = payload;
try {
const order = JSON.parse(value.toString());
if (!order.id || !order.customerId || !order.amount) {
throw new Error('Invalid order schema');
}
await this.processOrder(order);
} catch (error) {
console.error(`Poison pill at offset ${offset}:`, error.message);
await this.sendToDLQ(topic, partition, offset, timestamp, value.toString(), error);
}
// Commit the offset regardless of success or failure
// This moves the consumer forward past the poison pill
await payload.heartbeat();
}
async subscribe(): Promise {
await this.consumer.connect();
await this.dlqProducer.connect();
await this.consumer.subscribe({ topic: 'orders' });
await this.consumer.run({
eachMessage: this.handleOrderCreated.bind(this),
autoCommit: true, // Or handle commits manually in your logic
});
}
The trick is to commit after both the success and error paths. Your consumer offset advances, the poison pill is left behind in the DLQ, and your main consumer keeps processing fresh messages.
Alert when the DLQ grows
A DLQ that silently accumulates poison pills is worse than useless—it lulls you into a false sense of security. You need to monitor it and alert when messages start piling up.
Here's a monitoring pattern using a scheduled job:
import { Injectable, Logger } from '@nestjs/common';
import { Cron } from '@nestjs/schedule';
import { Kafka } from 'kafkajs';
@Injectable()
export class DLQMonitor {
private readonly logger = new Logger(DLQMonitor.name);
private dlqConsumer = this.kafka.consumer({ groupId: 'dlq-monitor' });
private dlqMessageCount = 0;
constructor(private kafka: Kafka) {}
@Cron('0 */5 * * * *') // Run every 5 minutes
async monitorDLQ(): Promise {
try {
await this.dlqConsumer.connect();
await this.dlqConsumer.subscribe({ topic: 'orders-dlq' });
let messageCount = 0;
await this.dlqConsumer.run({
eachMessage: async () => {
messageCount++;
},
autoCommitInterval: null, // Disable auto-commit for monitoring
});
// After consuming, check the count
if (messageCount > 0) {
this.logger.warn(`DLQ has ${messageCount} messages`);
// Alert if threshold exceeded
if (messageCount > 100) {
await this.sendAlert(
`CRITICAL: orders-dlq has ${messageCount} poison pills`,
'critical'
);
} else if (messageCount > 10) {
await this.sendAlert(
`WARNING: orders-dlq has ${messageCount} poison pills`,
'warning'
);
}
}
await this.dlqConsumer.disconnect();
} catch (error) {
this.logger.error('DLQ monitoring failed:', error);
}
}
private async sendAlert(message: string, severity: 'warning' | 'critical'): Promise {
// Send to your alert system: Slack, PagerDuty, CloudWatch, etc.
// Example: POST to Slack webhook
const slackMessage = {
text: `[${severity.toUpperCase()}] ${message}`,
channel: '#alerts',
};
console.log('Alert:', slackMessage);
// In production: await fetch(SLACK_WEBHOOK, { method: 'POST', body: JSON.stringify(slackMessage) });
}
}
A better approach uses Kafka metrics directly. Most Kafka brokers expose consumer lag metrics; a high lag on your DLQ topic is a red flag.
Common mistakes
- Rethrowing after catch: If you rethrow the exception, your consumer crashes or loops forever. Catch it, log it, route it, and move on.
- Forgetting to commit: If you don't advance the offset, the same poison pill gets replayed eternally. Always commit, even on errors.
- Not monitoring the DLQ: A DLQ that grows silently is a ticking time bomb. Set up alerts on message count or consumer lag.
- Losing the original payload: Store the raw message bytes in the DLQ, not just the error. You'll need them to debug or replay later.
- Treating all errors the same: Deserialization errors (poison pills) are different from transient errors (database down). Only poison pills go to the DLQ; transient errors should retry.
Key takeaways
- Wrap deserialization in explicit try-catch blocks; don't let exceptions bubble up to Kafka's retry mechanism.
- Route malformed messages to a separate Dead Letter Queue with metadata about why they failed.
- Always commit the offset after processing, even if you've only quarantined the message to the DLQ.
- Monitor your DLQ actively—set alerts when message count or consumer lag exceeds safe thresholds.
- Store the raw payload and full error context in the DLQ so you can investigate and replay selectively later.
- A single poison pill can paralyze your entire pipeline if you don't handle it; this pattern is as critical as your primary business logic.
Comments
Loading comments…