Apache Kafka: Building Real-Time Data Pipelines

Apache Kafka is a distributed streaming platform for building real-time data pipelines. This guide covers essential patterns for producing, consuming, and processing streaming data.

Core Concepts#

┌─────────────────────────────────────────────────────────────┐
│                        KAFKA CLUSTER                         │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│    ┌─────────┐    ┌─────────┐    ┌─────────┐               │
│    │ Broker  │    │ Broker  │    │ Broker  │               │
│    │   1     │    │   2     │    │   3     │               │
│    └─────────┘    └─────────┘    └─────────┘               │
│                                                              │
│    Topic: orders                                             │
│    ┌────────────┬────────────┬────────────┐                 │
│    │ Partition 0│ Partition 1│ Partition 2│                 │
│    │  [0,1,2,3] │  [0,1,2]   │  [0,1,2,3,4]                │
│    └────────────┴────────────┴────────────┘                 │
└─────────────────────────────────────────────────────────────┘

Producers ─────────────────► Topics ─────────────────► Consumers

Producer Implementation#

Basic Producer#

import { Kafka, Partitioners } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092'],
  ssl: true,
  sasl: {
    mechanism: 'plain',
    username: process.env.KAFKA_USERNAME!,
    password: process.env.KAFKA_PASSWORD!,
  },
});

const producer = kafka.producer({
  createPartitioner: Partitioners.DefaultPartitioner,
  idempotent: true, // Exactly-once semantics
  maxInFlightRequests: 5,
  transactionTimeout: 60000,
});

async function initProducer() {
  await producer.connect();

  // Handle disconnects
  producer.on('producer.disconnect', () => {
    console.error('Producer disconnected');
    // Implement reconnection logic
  });
}

async function sendEvent(topic: string, event: object, key?: string) {
  try {
    const result = await producer.send({
      topic,
      messages: [
        {
          key: key || undefined,
          value: JSON.stringify(event),
          headers: {
            'correlation-id': crypto.randomUUID(),
            'event-type': event.type,
            'timestamp': Date.now().toString(),
          },
        },
      ],
    });

    console.log('Message sent:', result);
    return result;
  } catch (error) {
    console.error('Failed to send message:', error);
    throw error;
  }
}

Batched Producer#

class BatchProducer {
  private batch: Array<{ topic: string; messages: any[] }> = [];
  private batchSize = 100;
  private flushInterval = 1000;
  private timer: NodeJS.Timeout | null = null;

  constructor(private producer: Producer) {
    this.startFlushTimer();
  }

  async send(topic: string, event: object, key?: string) {
    const topicBatch = this.batch.find(b => b.topic === topic);

    const message = {
      key: key || undefined,
      value: JSON.stringify(event),
    };

    if (topicBatch) {
      topicBatch.messages.push(message);
    } else {
      this.batch.push({ topic, messages: [message] });
    }

    if (this.getTotalMessages() >= this.batchSize) {
      await this.flush();
    }
  }

  private getTotalMessages(): number {
    return this.batch.reduce((sum, b) => sum + b.messages.length, 0);
  }

  async flush() {
    if (this.batch.length === 0) return;

    const currentBatch = this.batch;
    this.batch = [];

    await this.producer.sendBatch({
      topicMessages: currentBatch,
    });
  }

  private startFlushTimer() {
    this.timer = setInterval(() => this.flush(), this.flushInterval);
  }

  async shutdown() {
    if (this.timer) clearInterval(this.timer);
    await this.flush();
  }
}

Consumer Implementation#

Basic Consumer#

const consumer = kafka.consumer({
  groupId: 'order-processor',
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
  maxBytesPerPartition: 1048576, // 1MB
  retry: {
    retries: 5,
    initialRetryTime: 100,
    maxRetryTime: 30000,
  },
});

async function startConsumer() {
  await consumer.connect();

  await consumer.subscribe({
    topics: ['orders', 'payments'],
    fromBeginning: false,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const event = JSON.parse(message.value!.toString());

      console.log({
        topic,
        partition,
        offset: message.offset,
        key: message.key?.toString(),
        event,
      });

      await processEvent(event);
    },
  });
}

// Graceful shutdown
process.on('SIGTERM', async () => {
  await consumer.disconnect();
  process.exit(0);
});

Batch Consumer#

await consumer.run({
  eachBatch: async ({
    batch,
    resolveOffset,
    heartbeat,
    commitOffsetsIfNecessary,
    isRunning,
    isStale,
  }) => {
    for (const message of batch.messages) {
      if (!isRunning() || isStale()) break;

      try {
        const event = JSON.parse(message.value!.toString());
        await processEvent(event);

        resolveOffset(message.offset);
        await heartbeat();
      } catch (error) {
        console.error('Processing error:', error);
        // Don't resolve offset - will retry
        throw error;
      }
    }

    await commitOffsetsIfNecessary();
  },
});

Stream Processing#

Kafka Streams with TypeScript#

import { KafkaStreams } from 'kafka-streams';

const config = {
  kafkaHost: 'kafka:9092',
  groupId: 'stream-processor',
  clientName: 'analytics-stream',
};

const kafkaStreams = new KafkaStreams(config);

// Stream processing pipeline
const stream = kafkaStreams.getKStream('page-views');

stream
  // Parse JSON
  .map((message) => JSON.parse(message.value))

  // Filter by event type
  .filter((event) => event.type === 'page_view')

  // Group by page
  .branch([
    (event) => event.page.startsWith('/products'),
    (event) => event.page.startsWith('/checkout'),
  ])
  .map((branches) => {
    const [productViews, checkoutViews] = branches;

    productViews
      .countByKey('product_id', 'tumbling', 60000) // 1 minute window
      .to('product-view-counts');

    checkoutViews
      .countByKey('user_id', 'tumbling', 300000) // 5 minute window
      .to('checkout-funnel');
  });

stream.start();

Event Aggregation#

interface PageViewEvent {
  userId: string;
  page: string;
  timestamp: number;
}

interface SessionAggregate {
  userId: string;
  pages: string[];
  startTime: number;
  endTime: number;
  duration: number;
}

class SessionAggregator {
  private sessions: Map<string, SessionAggregate> = new Map();
  private sessionTimeout = 30 * 60 * 1000; // 30 minutes

  process(event: PageViewEvent): SessionAggregate | null {
    const existing = this.sessions.get(event.userId);

    if (existing) {
      // Check if session expired
      if (event.timestamp - existing.endTime > this.sessionTimeout) {
        // Emit completed session
        this.sessions.delete(event.userId);
        this.startNewSession(event);
        return existing;
      }

      // Update existing session
      existing.pages.push(event.page);
      existing.endTime = event.timestamp;
      existing.duration = existing.endTime - existing.startTime;
      return null;
    }

    this.startNewSession(event);
    return null;
  }

  private startNewSession(event: PageViewEvent) {
    this.sessions.set(event.userId, {
      userId: event.userId,
      pages: [event.page],
      startTime: event.timestamp,
      endTime: event.timestamp,
      duration: 0,
    });
  }
}

Error Handling#

Dead Letter Queue#

async function processWithDLQ(message: KafkaMessage, topic: string) {
  const retryCount = parseInt(
    message.headers?.['retry-count']?.toString() || '0'
  );

  try {
    const event = JSON.parse(message.value!.toString());
    await processEvent(event);
  } catch (error) {
    if (retryCount < 3) {
      // Retry with exponential backoff
      await producer.send({
        topic: `${topic}-retry`,
        messages: [{
          key: message.key,
          value: message.value,
          headers: {
            ...message.headers,
            'retry-count': (retryCount + 1).toString(),
            'original-topic': topic,
            'error': error.message,
          },
        }],
      });
    } else {
      // Send to DLQ
      await producer.send({
        topic: `${topic}-dlq`,
        messages: [{
          key: message.key,
          value: message.value,
          headers: {
            ...message.headers,
            'failed-at': new Date().toISOString(),
            'error': error.message,
            'stack': error.stack,
          },
        }],
      });
    }
  }
}

Consumer Error Recovery#

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    try {
      await processMessage(message);
    } catch (error) {
      if (isRetryableError(error)) {
        // Pause partition and retry after delay
        consumer.pause([{ topic, partitions: [partition] }]);

        setTimeout(() => {
          consumer.resume([{ topic, partitions: [partition] }]);
        }, 5000);

        throw error; // Don't commit offset
      }

      // Non-retryable error - send to DLQ
      await sendToDLQ(topic, message, error);
    }
  },
});

function isRetryableError(error: Error): boolean {
  return (
    error.message.includes('timeout') ||
    error.message.includes('connection') ||
    error.message.includes('temporarily unavailable')
  );
}

Monitoring#

Consumer Lag Monitoring#

const admin = kafka.admin();

async function getConsumerLag(groupId: string) {
  const groupDescription = await admin.describeGroups([groupId]);
  const topics = await admin.fetchTopicOffsets(groupId);

  const lag: Record<string, number> = {};

  for (const [topic, partitions] of Object.entries(topics)) {
    const topicOffsets = await admin.fetchTopicOffsets(topic);

    let topicLag = 0;
    for (const partition of partitions) {
      const latestOffset = topicOffsets.find(
        (o) => o.partition === partition.partition
      );
      if (latestOffset) {
        topicLag += parseInt(latestOffset.offset) - parseInt(partition.offset);
      }
    }
    lag[topic] = topicLag;
  }

  return lag;
}

// Expose as Prometheus metrics
import { Gauge } from 'prom-client';

const consumerLag = new Gauge({
  name: 'kafka_consumer_lag',
  help: 'Consumer group lag',
  labelNames: ['topic', 'group'],
});

setInterval(async () => {
  const lag = await getConsumerLag('order-processor');
  for (const [topic, value] of Object.entries(lag)) {
    consumerLag.set({ topic, group: 'order-processor' }, value);
  }
}, 10000);

Best Practices#

Use meaningful keys: For proper partitioning and ordering
Set appropriate retention: Balance storage with replay needs
Monitor consumer lag: Alert on growing lag
Implement idempotency: Handle duplicate messages
Use schemas: Avro or Protobuf for type safety
Test failure scenarios: Network issues, broker failures

Kafka enables scalable, fault-tolerant data streaming. Start with simple producers and consumers, then add stream processing as complexity grows. Focus on proper error handling and monitoring for production reliability.