Real-Time Data Pipelines with Apache Kafka and TypeScript
Build real-time data pipelines with Apache Kafka and TypeScript: partitioning strategies, consumer groups, exactly-once semantics and dead letter queues.
Why Kafka for Real-Time Pipelines
Batch processing has its place, but modern applications need to react to events as they happen—fraud detection, real-time analytics, inventory updates, notification systems. Kafka provides a distributed, durable, high-throughput event streaming platform that decouples producers from consumers and guarantees message ordering within partitions.
Producing Events
Producers send events to Kafka topics. Each event has a key that determines which partition it lands in. Events with the same key always go to the same partition, guaranteeing ordering for that key.
import { Kafka, Partitioners } from "kafkajs";
const kafka = new Kafka({
clientId: "order-service",
brokers: ["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"],
});
const producer = kafka.producer({
createPartitioner: Partitioners.DefaultPartitioner,
idempotent: true, // Prevent duplicate messages on retry
maxInFlightRequests: 5,
retry: { retries: 5 },
});
interface OrderEvent {
eventType: "order.created" | "order.updated" | "order.cancelled";
orderId: string;
userId: string;
payload: Record<string, unknown>;
timestamp: string;
}
async function publishOrderEvent(event: OrderEvent): Promise<void> {
await producer.send({
topic: "orders",
messages: [
{
// Key = orderId ensures all events for an order go to same partition
key: event.orderId,
value: JSON.stringify(event),
headers: {
"event-type": event.eventType,
"correlation-id": crypto.randomUUID(),
},
},
],
acks: -1, // Wait for all replicas to acknowledge
});
}// ❌ No key — events for same order scattered across partitions
await producer.send({
topic: "orders",
messages: [{ value: JSON.stringify(event) }], // Random partition
});
// ✅ Key-based partitioning preserves ordering per entity
await producer.send({
topic: "orders",
messages: [{
key: event.orderId, // Same order always same partition
value: JSON.stringify(event),
}],
});Consumer Groups and Parallel Processing
Consumers in the same group split partitions between them. If a topic has 12 partitions and 4 consumers in a group, each consumer processes 3 partitions. Adding consumers automatically rebalances the workload.
const consumer = kafka.consumer({
groupId: "analytics-pipeline",
sessionTimeout: 30_000,
heartbeatInterval: 3_000,
maxWaitTimeInMs: 100,
});
async function startConsumer(): Promise<void> {
await consumer.connect();
await consumer.subscribe({
topics: ["orders"],
fromBeginning: false,
});
await consumer.run({
autoCommit: false, // Manual commit for exactly-once processing
eachMessage: async ({ topic, partition, message }) => {
const event: OrderEvent = JSON.parse(message.value!.toString());
try {
await processEvent(event);
// Commit only after successful processing
await consumer.commitOffsets([
{
topic,
partition,
offset: (Number(message.offset) + 1).toString(),
},
]);
} catch (error) {
await handleProcessingError(event, error as Error);
}
},
});
}Dead Letter Queues for Failed Messages
Some messages cannot be processed—invalid data, missing references, transient bugs. Instead of blocking the partition or losing the message, route failures to a dead letter queue for investigation and reprocessing.
interface DeadLetterMessage {
originalTopic: string;
originalPartition: number;
originalOffset: string;
originalKey: string | null;
originalValue: string;
errorMessage: string;
errorStack: string;
failedAt: string;
retryCount: number;
}
async function handleProcessingError(
event: OrderEvent,
error: Error,
context: { topic: string; partition: number; offset: string }
): Promise<void> {
const deadLetter: DeadLetterMessage = {
originalTopic: context.topic,
originalPartition: context.partition,
originalOffset: context.offset,
originalKey: event.orderId,
originalValue: JSON.stringify(event),
errorMessage: error.message,
errorStack: error.stack ?? "",
failedAt: new Date().toISOString(),
retryCount: 0,
};
await producer.send({
topic: `${context.topic}.dead-letter`,
messages: [
{
key: event.orderId,
value: JSON.stringify(deadLetter),
},
],
});
console.error(
`Event ${event.orderId} sent to dead letter queue: ${error.message}`
);
}Exactly-Once Processing with Transactions
Kafka supports transactions that atomically produce messages and commit consumer offsets. This achieves exactly-once semantics: each message is processed exactly once even across failures.
const transactionalProducer = kafka.producer({
idempotent: true,
transactionalId: "analytics-transformer",
maxInFlightRequests: 1,
});
async function processWithTransaction(
messages: Array<{ topic: string; partition: number; message: KafkaMessage }>
): Promise<void> {
const transaction = await transactionalProducer.transaction();
try {
for (const { topic, partition, message } of messages) {
const event: OrderEvent = JSON.parse(message.value!.toString());
// Transform and produce to downstream topic
const enriched = await enrichEvent(event);
await transaction.send({
topic: "orders-enriched",
messages: [
{
key: event.orderId,
value: JSON.stringify(enriched),
},
],
});
// Commit consumer offset within the transaction
await transaction.sendOffsets({
consumerGroupId: "analytics-pipeline",
topics: [
{
topic,
partitions: [
{
partition,
offset: (Number(message.offset) + 1).toString(),
},
],
},
],
});
}
await transaction.commit();
} catch (error) {
await transaction.abort();
throw error;
}
}Schema Evolution with Versioned Events
As your domain evolves, event schemas change. Use a schema registry or embedded versioning to handle backward and forward compatibility.
interface VersionedEvent<T> {
schemaVersion: number;
eventType: string;
data: T;
}
// Version-aware deserializer
type EventHandler<T> = (data: T) => Promise<void>;
class EventDeserializer {
private handlers: Map<string, Map<number, EventHandler<unknown>>> = new Map();
register<T>(
eventType: string,
version: number,
handler: EventHandler<T>
): void {
if (!this.handlers.has(eventType)) {
this.handlers.set(eventType, new Map());
}
this.handlers.get(eventType)!.set(version, handler as EventHandler<unknown>);
}
async handle(raw: string): Promise<void> {
const event: VersionedEvent<unknown> = JSON.parse(raw);
const versionHandlers = this.handlers.get(event.eventType);
if (!versionHandlers) {
throw new Error(`Unknown event type: ${event.eventType}`);
}
const handler = versionHandlers.get(event.schemaVersion);
if (!handler) {
// Try to find the latest version handler that can upcast
const latest = Math.max(...versionHandlers.keys());
const latestHandler = versionHandlers.get(latest);
if (latestHandler) {
const upcasted = upcastEvent(event, latest);
await latestHandler(upcasted.data);
return;
}
throw new Error(
`No handler for ${event.eventType} v${event.schemaVersion}`
);
}
await handler(event.data);
}
}Key Takeaways
Kafka pipelines decouple producers from consumers, enabling real-time processing at scale. Use message keys to guarantee ordering per entity—all events for an order go to the same partition. Consumer groups parallelize processing automatically; add consumers to scale horizontally.
Disable auto-commit and manage offsets manually for reliable processing. Route failed messages to dead letter queues instead of blocking partitions or losing data. Use Kafka transactions for exactly-once semantics when transforming events between topics. Version your event schemas from day one—breaking changes in event formats are painful to fix after the fact. Start simple with a producer, a consumer group, and a dead letter queue; add transactions and schema registries as your pipeline matures.
