Claude Code with Kafka: Event Streams That Do Not Lose Messages
Why Kafka without CLAUDE.md ships consumers that quietly drop events
Kafka is the most powerful piece of streaming infrastructure most teams will ever run. It is also the most ways to lose data without realising. The defaults of every Kafka client library are tuned for throughput. The defaults of production are tuned for durability. Bridging the gap is the entire content of every Kafka book ever written, and it is what Claude Code does not do without a CLAUDE.md that pins the right defaults.
Asked to "write a Kafka consumer in Node," Claude generates a consumer that auto-commits offsets every five seconds, processes messages without idempotency keys, hardcodes the consumer group as 'my-group', parses payloads with JSON.parse and no schema check, and has no error handling for messages it cannot process. Each of those defaults is a way to lose or duplicate messages. The auto-commit means a crash mid-batch loses every message processed since the last commit. The lack of idempotency means a rebalance reprocesses messages and side-effects fire twice. The hardcoded group name means two services with the same code accidentally share a consumer group and steal each other's messages. The missing schema check means a publisher change downstream silently corrupts every consumer.
This guide covers the CLAUDE.md configuration that locks Claude Code into Kafka patterns that hold up: idempotent producers with the right acks setting, manual offset commits after batch processing, partition keys derived from business identifiers, consumer group naming that encodes service identity, schema registry contracts on every topic, and the dead-letter pattern for messages the consumer cannot handle. For the broker operational concerns (replication, partitioning at scale, monitoring), Claude Code with Datadog covers the observability layer. For the alternative messaging model where ordered streams are overkill, Claude Code with BullMQ covers the job-queue pattern that handles request/response style work more cleanly.
The Kafka CLAUDE.md template
The CLAUDE.md at your project root is read at the start of every Claude Code session. For a Kafka integration it needs to declare: the client library, the broker connection, the producer durability settings, the consumer group conventions, the offset commit strategy, the schema contract, the dead-letter routing, and the hard rules that block the data-loss patterns Claude generates without them.
# Kafka rules
## Stack
- kafkajs ^2.x (Node.js) or confluent-kafka-go for Go services
- Schema Registry: Confluent Schema Registry or Karapace, Avro or Protobuf payloads
- KAFKA_BROKERS env var (comma-separated host:port list), no inline hosts
- SCHEMA_REGISTRY_URL env var
- One service identity per repo: KAFKA_CLIENT_ID = '<service>-<env>'
## Project structure
- src/kafka/ , shared client init (producer, consumer factory)
- src/kafka/topics.ts , topic constants + partition key contracts
- src/kafka/schemas/ , Avro/Protobuf schema files (versioned)
- src/producers/ , one file per producer (typed, async)
- src/consumers/ , one file per consumer group (consumer class)
- src/handlers/ , pure functions called by consumers (testable)
- src/dlq/ , dead-letter handling + replay tooling
- tests/ , unit tests for handlers + integration tests with real broker
## Producer settings (MANDATORY)
- idempotent: true (exactly-once produce semantics)
- acks: -1 (all in-sync replicas must acknowledge)
- retries: 5 minimum, 10 default
- maxInFlightRequests: 5 with idempotent (preserves ordering)
- compression: 'gzip' or 'lz4' for high-throughput topics
## Partition keys (MANDATORY for every produce call)
- ALWAYS pass a key derived from the business identity (userId, accountId, orderId)
- NEVER produce without a key on a topic where ordering matters
- Key serialization: stable string, NOT JSON of an object (object key ordering varies)
- Partition count: 3x the maximum consumer concurrency for the topic
## Consumer group conventions (MANDATORY)
- groupId: '<service>-<purpose>-<env>' (e.g., 'billing-charge-processor-prod')
- NEVER reuse a groupId across services
- NEVER use 'default', 'my-group', 'test' in production
- groupId is documented in src/kafka/topics.ts alongside the topic it consumes
## Offset commit strategy (MANDATORY)
- autoCommit: false on every consumer in production
- Commit AFTER handler completes successfully for a batch
- Commit interval: per batch, NOT per message (avoid commit-per-message overhead)
- On handler failure: do NOT commit, the message is redelivered
## Schema contracts (MANDATORY)
- Every topic has an Avro or Protobuf schema registered in Schema Registry
- Producer serializes via @kafkajs/confluent-schema-registry (or equivalent)
- Consumer deserializes via the same library, FAIL FAST on schema mismatch
- Schema evolution: backward-compatible additions only without a major version bump
## Idempotency in consumers (MANDATORY)
- Every consumer handler is idempotent (deduped by business key)
- Idempotency table or Redis SET to track processed message IDs
- TTL on the idempotency record matches the maximum replay window
## Dead letter handling
- Messages that fail the handler 3 times go to '<topic>-dlq'
- DLQ messages carry: original key, original payload, failure reason, attempt count
- DLQ has its own consumer for manual replay or human investigation
## Hard rules
- NEVER autoCommit in a production consumer
- NEVER produce without a partition key on an ordered topic
- NEVER use JSON.stringify(obj) as a partition key
- NEVER reuse a consumer groupId across services
- NEVER skip Schema Registry for a topic in production
- NEVER catch-and-swallow errors in a consumer handler (let them propagate, retry, then DLQ)
- ALWAYS log topic + partition + offset + key on every consumer event
Six rules here prevent the majority of production incidents Claude generates without them.
The idempotent producer rule is the foundation for at-least-once produce. Without idempotent: true, a producer retry can write the same message twice. With it, the broker tracks producer sequence numbers and rejects duplicates. The cost is a slightly higher per-produce latency. The benefit is that retries are safe. The acks: -1 setting is the matching durability guarantee on the broker side: the produce only acknowledges once all in-sync replicas have written the record, so a broker failure does not lose in-flight messages.
The partition key rule is the single most violated rule in Kafka code. The partition key determines which partition a message lands in, and therefore which consumer handles it. Messages with the same key go to the same partition, which guarantees order. Messages with no key are spread randomly. A topic that needs per-user ordering (state updates, balance changes, account events) produced without a key has no ordering guarantee, and Claude will write the consumer code as if ordering is preserved. The right key is the business identity: userId for user events, orderId for order events, never the message ID and never a random UUID.
The manual offset commit rule is the most important consumer rule. Kafka's autoCommit commits offsets on a timer (default every five seconds), regardless of whether the handler has finished processing the message. A consumer that crashes between auto-commit and handler completion loses every message in the current batch. The right pattern is autoCommit: false, process the batch, commit the offset after the handler returns. If the consumer crashes mid-batch, the next consumer picks up from the last committed offset and reprocesses the batch.
The consumer group naming rule prevents the silent message-stealing class of bug. Two services with the same consumer group share the topic's partitions: each partition is assigned to exactly one consumer in the group. If service A and service B both use groupId: 'event-processor', half the messages go to A and half go to B. Each service thinks it is processing all messages. Both are wrong. The naming convention <service>-<purpose>-<env> makes accidental collisions impossible.
The schema contract rule prevents the silent corruption class of bug. JSON over Kafka is a foot-gun: a publisher adds a field, every consumer keeps working, then someone reads the field expecting it to be there and gets undefined. Avro or Protobuf with a Schema Registry enforces a contract. The publisher cannot register a backward-incompatible schema. The consumer fails fast on a schema mismatch instead of silently processing wrong data.
Install and client setup
Install KafkaJS and the Schema Registry client:
pnpm add kafkajs @kafkajs/confluent-schema-registry
pnpm add -D @types/node
Set the environment:
# .env.local
KAFKA_BROKERS=localhost:9092
SCHEMA_REGISTRY_URL=http://localhost:8081
KAFKA_CLIENT_ID=billing-local
Create the shared Kafka client:
// src/kafka/client.ts
import { Kafka, logLevel } from 'kafkajs';
if (!process.env.KAFKA_BROKERS) {
throw new Error('KAFKA_BROKERS must be set');
}
export const kafka = new Kafka({
clientId: process.env.KAFKA_CLIENT_ID ?? 'unset-client',
brokers: process.env.KAFKA_BROKERS.split(','),
logLevel: logLevel.WARN,
retry: {
initialRetryTime: 300,
retries: 8,
},
});
The clientId is what Kafka uses to identify the connection in broker logs and metrics. If two services share a clientId, debugging becomes painful. Enforce a one-clientId-per-service convention via the env var, and have the deployment manifest set it explicitly.
Create the Schema Registry client:
// src/kafka/schema-registry.ts
import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
if (!process.env.SCHEMA_REGISTRY_URL) {
throw new Error('SCHEMA_REGISTRY_URL must be set');
}
export const schemaRegistry = new SchemaRegistry({
host: process.env.SCHEMA_REGISTRY_URL,
});
Both modules are singletons and imported wherever a producer or consumer is constructed.
Defining a topic and schema
Topic names, partition counts, and schemas are application contracts. Document them in src/kafka/topics.ts with the partition key strategy and the consuming services.
// src/kafka/topics.ts
export const ORDER_PLACED = {
name: 'order.placed.v1',
partitions: 24,
replication: 3,
partitionKey: 'orderId',
schemaSubject: 'order.placed.v1-value',
consumers: ['billing-charge-processor', 'inventory-reservation', 'notification-sender'],
} as const;
export const PAYMENT_COMPLETED = {
name: 'payment.completed.v1',
partitions: 12,
replication: 3,
partitionKey: 'paymentId',
schemaSubject: 'payment.completed.v1-value',
consumers: ['order-state-machine', 'analytics-events'],
} as const;
The partitionKey field documents what the producer must pass as the key. The consumers array is a register of who consumes the topic, which prevents the "let's just consume this topic from a new service" pattern that accumulates over time without anyone tracking.
The Avro schema lives in src/kafka/schemas/:
{
"type": "record",
"name": "OrderPlaced",
"namespace": "com.example.order",
"fields": [
{ "name": "orderId", "type": "string" },
{ "name": "userId", "type": "string" },
{ "name": "amount", "type": "long", "doc": "amount in minor currency units" },
{ "name": "currency", "type": "string", "doc": "ISO 4217" },
{ "name": "items", "type": { "type": "array", "items": "string" } },
{ "name": "placedAt", "type": { "type": "long", "logicalType": "timestamp-millis" } }
]
}
Register the schema once via the Schema Registry API or your deployment pipeline. Schema evolution rule: adding optional fields with defaults is backward-compatible. Removing fields, renaming fields, or changing types is not.
Writing an idempotent producer
The producer initializes once per process and is reused across produce calls. Wrap it in a typed function that enforces the partition key and the schema.
// src/producers/order-placed.ts
import { kafka } from '../kafka/client';
import { schemaRegistry } from '../kafka/schema-registry';
import { ORDER_PLACED } from '../kafka/topics';
const producer = kafka.producer({
idempotent: true,
maxInFlightRequests: 5,
allowAutoTopicCreation: false,
});
let connected = false;
async function ensureConnected() {
if (!connected) {
await producer.connect();
connected = true;
}
}
export type OrderPlacedEvent = {
orderId: string;
userId: string;
amount: number;
currency: string;
items: string[];
placedAt: number;
};
export async function publishOrderPlaced(event: OrderPlacedEvent) {
await ensureConnected();
const value = await schemaRegistry.encode(
await schemaRegistry.getRegistryIdBySchema(ORDER_PLACED.schemaSubject, schemaJson),
event,
);
await producer.send({
topic: ORDER_PLACED.name,
acks: -1,
messages: [
{
key: event.orderId,
value,
headers: {
'event-type': 'order.placed.v1',
'producer-service': process.env.KAFKA_CLIENT_ID ?? '',
},
},
],
});
}
Three things CLAUDE.md enforces here. The producer is constructed once at module load with idempotent: true, so retries on transient errors do not double-write. The send call passes key: event.orderId, so all events for the same order land in the same partition and are processed in order. The schema is fetched from the registry and used to encode the value, so a publisher cannot accidentally produce a payload that violates the contract.
The allowAutoTopicCreation: false setting prevents the typo class of bug. A producer that targets order.place.v1 (missing a "d") with auto-create enabled spawns a new, unmonitored topic. With auto-create disabled, the produce fails loudly and someone fixes the typo.
Writing a consumer with manual offset commits
The consumer is where most of the data-loss patterns live. Set autoCommit: false, commit after the batch, and handle errors deliberately.
// src/consumers/charge-processor.ts
import { kafka } from '../kafka/client';
import { schemaRegistry } from '../kafka/schema-registry';
import { ORDER_PLACED } from '../kafka/topics';
import { processCharge } from '../handlers/process-charge';
const consumer = kafka.consumer({
groupId: 'billing-charge-processor-prod',
sessionTimeout: 30000,
heartbeatInterval: 3000,
rebalanceTimeout: 60000,
maxBytesPerPartition: 1048576,
});
export async function startChargeProcessor() {
await consumer.connect();
await consumer.subscribe({
topic: ORDER_PLACED.name,
fromBeginning: false,
});
await consumer.run({
autoCommit: false,
partitionsConsumedConcurrently: 4,
eachBatch: async ({ batch, resolveOffset, heartbeat, commitOffsetsIfNecessary }) => {
for (const message of batch.messages) {
if (!message.value) continue;
const event = await schemaRegistry.decode(message.value);
try {
await processCharge(event, {
messageId: `${batch.topic}:${batch.partition}:${message.offset}`,
});
resolveOffset(message.offset);
} catch (err) {
console.error('charge_processor_error', {
topic: batch.topic,
partition: batch.partition,
offset: message.offset,
key: message.key?.toString(),
error: (err as Error).message,
});
throw err;
}
await heartbeat();
}
await commitOffsetsIfNecessary();
},
});
}
The eachBatch handler iterates messages in the batch, decodes via the schema registry, calls the handler with the message ID, and only calls resolveOffset after the handler completes. The throw err re-raises any handler failure, which causes KafkaJS to abort the batch without committing the offset. The next poll picks up the same offset and the message is reprocessed. The heartbeat() call inside the loop keeps the broker from marking the consumer as failed during long batches.
The handler in src/handlers/process-charge.ts is the idempotency boundary:
// src/handlers/process-charge.ts
import { redis } from '../redis/connection';
import { chargeCustomer } from '../billing/charge';
const IDEMPOTENCY_TTL_SECONDS = 60 * 60 * 24 * 7;
export async function processCharge(
event: { orderId: string; amount: number; currency: string },
context: { messageId: string },
) {
const key = `charge-processor:processed:${event.orderId}`;
const wasFirstTime = await redis.set(key, context.messageId, 'EX', IDEMPOTENCY_TTL_SECONDS, 'NX');
if (wasFirstTime !== 'OK') {
return { status: 'already-processed', orderId: event.orderId };
}
const result = await chargeCustomer({
orderId: event.orderId,
amount: event.amount,
currency: event.currency,
});
return { status: 'charged', chargeId: result.chargeId };
}
The Redis SET NX EX is the idempotency anchor. The first time the handler runs for an orderId, the SET succeeds and the handler charges the customer. The second time (after a rebalance, after a crash-and-redelivery), the SET fails and the handler returns early. The Redis TTL gives a bounded replay window after which the same orderId can be processed again, which is rare but matters for replays from beginning-of-time.
Dead letter routing
Some messages should not be retried forever. A schema-deserialization failure means the message is corrupt; retrying will not help. A handler exception that the code knows is non-retryable should route to a DLQ instead of blocking the partition.
// inside the eachBatch handler, replace the throw err block with:
} catch (err) {
if (isNonRetryable(err)) {
await dlqProducer.send({
topic: `${batch.topic}-dlq`,
messages: [
{
key: message.key,
value: message.value,
headers: {
'original-topic': batch.topic,
'failure-reason': (err as Error).message,
'attempt-count': '1',
'failed-at': new Date().toISOString(),
},
},
],
});
resolveOffset(message.offset);
} else {
throw err;
}
}
Non-retryable errors are routed to the DLQ topic and the offset is committed, so the partition progresses. Retryable errors fall through to the throw err path and the message gets reprocessed on the next poll. The classification is the application's responsibility: deserialization errors, schema mismatch, bad references to deleted entities are non-retryable; network errors, upstream timeouts, transient broker errors are retryable.
A dedicated DLQ consumer routes messages from the DLQ topic to operator alerts and provides a replay UI for manual recovery. The DLQ topic uses the same Schema Registry contract as the original topic, plus headers that carry the failure context.
Consumer rebalances and graceful shutdown
A consumer that exits without leaving the group cleanly causes a rebalance. The other consumers pause while partitions are reassigned. Long rebalance times mean lag spikes and missed SLAs. Graceful shutdown sends a LeaveGroup request before exiting, which keeps rebalances fast.
// src/main.ts
import { startChargeProcessor } from './consumers/charge-processor';
const consumer = await startChargeProcessor();
async function shutdown(signal: string) {
console.log('shutdown_start', { signal });
await consumer.disconnect();
console.log('shutdown_complete');
process.exit(0);
}
process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));
consumer.disconnect() sends the LeaveGroup request and resolves once the broker acknowledges. Match terminationGracePeriodSeconds in your Kubernetes manifest to the consumer's sessionTimeout plus a buffer. If sessionTimeout is 30s and the grace period is 10s, the orchestrator kills the process before it can leave the group cleanly, and the rebalance takes the full session timeout.
Common Claude Code mistakes with Kafka
Six patterns Claude generates incorrectly without CLAUDE.md constraints.
1. Producer without idempotent: true
Claude generates: kafka.producer().
Correct pattern: kafka.producer({ idempotent: true, maxInFlightRequests: 5 }). Retries are now safe.
2. Produce without a partition key
Claude generates: producer.send({ topic, messages: [{ value: serialize(event) }] }).
Correct pattern: producer.send({ topic, messages: [{ key: event.orderId, value: serialize(event) }] }). Ordering is now guaranteed per key.
3. Consumer with autoCommit
Claude generates: consumer.run({ eachMessage: async ({ message }) => process(message) }) with default autoCommit: true.
Correct pattern: consumer.run({ autoCommit: false, eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary }) => { ... } }).
4. Generic groupId
Claude generates: kafka.consumer({ groupId: 'my-group' }).
Correct pattern: kafka.consumer({ groupId: 'billing-charge-processor-prod' }). Service identity is now explicit.
5. JSON.parse without schema validation
Claude generates: const event = JSON.parse(message.value.toString()).
Correct pattern: const event = await schemaRegistry.decode(message.value). Schema contract is enforced.
6. Catch-and-continue on handler errors
Claude generates: try { await handle(message) } catch { /* log and continue */ }.
Correct pattern: throw the error so the batch is not committed; the message is redelivered. Handler logic is responsible for routing non-retryable errors to the DLQ.
Add each pattern to CLAUDE.md with the corrected form.
Permission hooks for Kafka CLI
Kafka has a set of admin operations that are dangerous in production. Topic deletion, partition reassignment, ACL changes, and consumer group reset can take a system down or lose data. Permission hooks gate them.
In .claude/settings.local.json:
{
"permissions": {
"allow": [
"Bash(kafka-topics --describe*)",
"Bash(kafka-topics --list*)",
"Bash(kafka-consumer-groups --describe*)",
"Bash(kafka-consumer-groups --list*)",
"Bash(kafka-console-consumer*)"
],
"deny": [
"Bash(kafka-topics --delete*)",
"Bash(kafka-consumer-groups --reset-offsets*)",
"Bash(kafka-consumer-groups --delete*)",
"Bash(kafka-acls --remove*)",
"Bash(kafka-reassign-partitions*)"
]
}
}
The allow list permits the read-only inspection commands. The deny list forces destructive operations into the explicit-confirmation path. Combined with the Claude Code hooks system for project-wide policy, the result is a setup where Claude can debug a stuck consumer but cannot delete the topic.
Observability
Every Kafka deployment needs three sets of metrics: consumer lag per group per partition, producer metrics (success rate, batch size, latency), and broker metrics (in-sync replica count, under-replicated partitions, request handler idle).
Consumer lag is the most important production signal. It is the difference between the latest offset on the partition and the consumer group's committed offset. A growing lag means the consumer is slower than the producer. Alert at lag > 60_000 messages or lag time > 5 minutes. Below that, lag is normal variation; above that, the consumer is falling behind.
Producer metrics tell you whether the producer is healthy. The success rate should be ~100%. Batch size tells you whether compression is helping. Latency p99 should be under the broker's request timeout.
Broker metrics are the cluster-health signal. Under-replicated partitions mean a broker is behind on replication, which is a precursor to data loss if the leader fails. In-sync replica count below replication.factor - 1 should page someone. For the observability stack itself, Claude Code with Datadog covers integrating Kafka metrics into a unified dashboard with alerting.
Schema evolution rules
A Kafka topic outlives the service that created it. The producer changes, the consumers change, the schema needs to evolve without breaking any of them. The rules for backward-compatible Avro evolution:
- Adding a field with a default value: safe.
- Removing a field with a default value: safe.
- Adding an optional field (union with
null): safe. - Renaming a field: unsafe. Use aliases instead.
- Changing a field type: unsafe in almost all cases.
- Changing a field's default value: unsafe.
The Schema Registry enforces a compatibility mode per subject. Set it to BACKWARD for topics where new consumers must read old data, or FULL for topics where producers and consumers can be deployed in any order. Document the mode in src/kafka/topics.ts alongside the topic.
Building Kafka code that does not lose messages
The Kafka CLAUDE.md in this guide produces code where every producer is idempotent and acks all replicas, every message has a partition key, every consumer commits offsets manually after batch processing, every handler is idempotent at the business-key level, every topic has a Schema Registry contract, and non-retryable failures route to a DLQ for human investigation.
The underlying principle is the same as any streaming infrastructure with Claude Code. Kafka without a CLAUDE.md produces code that works against a local broker, passes unit tests, and falls apart the first time a rebalance happens, a producer retries, or a publisher adds a field to a schema. The CLAUDE.md is what makes the production-safe defaults the default for Claude.
For the alternative messaging model where ordered streams are overkill and request/response semantics are the right shape, Claude Code with BullMQ covers job queues on Redis. For workflow orchestration where Kafka is too low-level, Claude Code with Temporal covers the workflow engine that handles state machines and sagas across streaming and request/response boundaries. For Redis itself, Claude Code with Redis covers the connection patterns the consumer idempotency layer depends on.
Get Claudify. Ship Kafka producers and consumers that survive a real broker.
More like this
Ready to upgrade your Claude Code setup?
Get Claudify
