Setting Up Apache Kafka for Node.js Message Queues

Introduction

Apache Kafka provides durable, scalable message streaming that handles millions of events per second. For Node.js applications, Kafka serves as the backbone of event-driven architectures, decoupling producers from consumers and enabling asynchronous processing patterns. However, configuring Kafka correctly requires understanding partitioning, consumer groups, and offset management.

After implementing Kafka in three production systems, I have documented the setup patterns that prevent the most common pitfalls, from message loss to consumer lag and rebalancing storms.

Environment

node --version
# v20.11.0

npm list kafkajs
# kafkajs@2.2.4

# Kafka broker version
docker exec kafka kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | head -1
# Kafka 3.6.1

The project structure:

kafka-service/
├── src/
│   ├── producers/
│   │   ├── event-producer.ts
│   │   └── batch-producer.ts
│   ├── consumers/
│   │   ├── order-consumer.ts
│   │   └── notification-consumer.ts
│   ├── config/
│   │   ├── kafka.ts
│   │   └── topics.ts
│   └── utils/
│       ├── retry.ts
│       └── metrics.ts
├── docker-compose.yml
└── package.json

Problem

The first issue was message loss during producer retries. When the Kafka broker was temporarily unavailable, the producer dropped messages instead of buffering them:

// Messages are lost when the broker is unreachable
const producer = kafka.producer();
await producer.connect();

await producer.send({
  topic: "orders",
  messages: [
    { key: "order-123", value: JSON.stringify(orderData) },
  ],
});
// If this fails, the message is lost forever

The second issue was consumer group rebalancing storms. When a consumer was slow processing a message, Kafka reassigned its partitions to other consumers, causing duplicate processing and further slowdown:

// Consumer logs show repeated rebalancing
[Consumer] Rebalance triggered: group instance expired
[Consumer] Rebalance triggered: member joined
[Consumer] Rebalance triggered: member left
// This cycle repeated every 30 seconds

The third issue was that the consumer could not distinguish between a poison message (which should be skipped) and a transient error (which should be retried). The dead letter queue was not configured, so poison messages caused infinite retry loops.

Analysis

Kafka producers have three delivery guarantees: acks=0 (fire and forget), acks=1 (leader acknowledges), and acks=all (all replicas acknowledge). The default acks=1 provides no guarantee that the message was replicated before the leader failed. For critical messages, acks=all with min.insync.replicas=2 ensures the message survives broker failures.

Consumer group rebalancing is triggered by the session.timeout.ms and heartbeat.interval.ms configurations. If a consumer takes longer than the session timeout to process a message, Kafka considers it dead and triggers a rebalance. The default session timeout of 30 seconds is too short for consumers that process complex messages.

The poison message problem occurs because Kafka consumers process messages sequentially within a partition. If a message causes a processing error, the consumer retries the same message indefinitely, blocking all subsequent messages in that partition. The dead letter queue pattern provides an escape hatch for poison messages.

Solution

Configure the producer with durability guarantees:

// producers/event-producer.ts
import { Kafka, Producer, ProducerRecord, Message } from "kafkajs";
import { kafka } from "../config/kafka";

class EventProducer {
  private producer: Producer;
  private buffer: Message[] = [];
  private flushInterval: NodeJS.Timer | null = null;

  constructor() {
    this.producer = kafka.producer({
      // Ensure all replicas acknowledge
      allowAutoTopicCreation: false,
      transactionTimeout: 30000,
    });
  }

  async connect() {
    await this.producer.connect();

    // Flush buffer every 5 seconds
    this.flushInterval = setInterval(() => {
      this.flushBuffer();
    }, 5000);
  }

  async send(topic: string, message: Message) {
    try {
      await this.producer.send({
        topic,
        messages: [message],
        acks: -1, // all replicas must acknowledge
        compression: 1, // gzip
        timeout: 10000,
      });
    } catch (error) {
      console.error(`Failed to send to ${topic}:`, error);
      this.buffer.push(message);
    }
  }

  private async flushBuffer() {
    if (this.buffer.length === 0) return;

    const messages = [...this.buffer];
    this.buffer = [];

    try {
      await this.producer.send({
        topic: "orders",
        messages,
        acks: -1,
      });
    } catch (error) {
      // Re-add failed messages to buffer
      this.buffer.unshift(...messages);
    }
  }

  async disconnect() {
    if (this.flushInterval) {
      clearInterval(this.flushInterval as NodeJS.Timeout);
    }
    await this.flushBuffer();
    await this.producer.disconnect();
  }
}

export const eventProducer = new EventProducer();

Configure consumers with proper timeouts and error handling:

// consumers/order-consumer.ts
import { Consumer, EachMessagePayload } from "kafkajs";
import { kafka } from "../config/kafka";

class OrderConsumer {
  private consumer: Consumer;

  constructor() {
    this.consumer = kafka.consumer({
      groupId: "order-processing-group",
      // Increase session timeout to prevent rebalancing
      sessionTimeout: 60000,
      heartbeatInterval: 15000,
      maxBytesPerPartition: 1048576, // 1MB
      minBytes: 1,
      maxBytes: 10485760, // 10MB
    });
  }

  async connect() {
    await this.consumer.connect();
    await this.consumer.subscribe({
      topics: ["orders", "payments"],
      fromBeginning: false,
    });

    await this.consumer.run({
      eachBatchAutoResolve: true,
      autoCommit: false,
      eachBatch: async ({
        batch,
        resolveOffset,
        heartbeat,
        commitOffsetsIfNecessary,
      }) => {
        for (const message of batch.messages) {
          try {
            await this.processMessage(batch.topic, message);
            resolveOffset(message.offset);
            await commitOffsetsIfNecessary({
              topic: batch.topic,
              partition: batch.partition,
              offset: (BigInt(message.offset) + 1n).toString(),
            });
          } catch (error) {
            await this.handleProcessingError(
              batch.topic,
              batch.partition,
              message,
              error as Error
            );
          }
          await heartbeat();
        }
      },
    });
  }

  private async processMessage(topic: string, message: any) {
    const data = JSON.parse(message.value!.toString());
    console.log(`Processing order: ${data.orderId}`);

    // Simulate processing
    await this.validateOrder(data);
    await this.processPayment(data);
    await this.updateInventory(data);

    console.log(`Order ${data.orderId} processed successfully`);
  }

  private async handleProcessingError(
    topic: string,
    partition: number,
    message: any,
    error: Error
  ) {
    console.error(`Processing error:`, error);

    // Send to dead letter queue
    await eventProducer.send(`${topic}.dlq`, {
      key: message.key,
      value: message.value,
      headers: {
        "original-topic": topic,
        "original-partition": partition.toString(),
        "original-offset": message.offset,
        "error-message": error.message,
        "retry-count": "0",
      },
    });
  }
}

export const orderConsumer = new OrderConsumer();

Configure Kafka topics with proper partitioning:

// config/topics.ts
export const topics = {
  orders: {
    topic: "orders",
    numPartitions: 6,
    replicationFactor: 3,
    configEntries: [
      { name: "retention.ms", value: "604800000" }, // 7 days
      { name: "cleanup.policy", value: "delete" },
      { name: "compression.type", value: "gzip" },
      { name: "max.message.bytes", value: "1048576" },
    ],
  },
  "orders.dlq": {
    topic: "orders.dlq",
    numPartitions: 3,
    replicationFactor: 3,
    configEntries: [
      { name: "retention.ms", value: "2592000000" }, // 30 days
    ],
  },
};

Implement consumer lag monitoring:

// utils/metrics.ts
import { Admin } from "kafkajs";
import { kafka } from "../config/kafka";

export async function getConsumerLag(): Promise<
  Array<{ topic: string; partition: number; lag: string }>
> {
  const admin: Admin = kafka.admin();
  await admin.connect();

  const groups = await admin.listGroups();
  const lagInfo: Array<{ topic: string; partition: number; lag: string }> =
    [];

  for (const group of groups.groups) {
    const offsets = await admin.fetchOffsets({
      groupId: group.groupId,
    });

    for (const offset of offsets) {
      const lag =
        BigInt(offset.high) - BigInt(offset.currentOffset);
      lagInfo.push({
        topic: offset.topic,
        partition: offset.partition,
        lag: lag.toString(),
      });
    }
  }

  await admin.disconnect();
  return lagInfo;
}

// Run periodically to detect consumer lag
setInterval(async () => {
  const lag = await getConsumerLag();
  const totalLag = lag.reduce((sum, item) => sum + BigInt(item.lag), BigInt(0));

  if (totalLag > BigInt(10000)) {
    console.warn(`Consumer lag detected: ${totalLag} messages`);
  }
}, 30000);

Set up the Docker Compose environment:

# docker-compose.yml
version: "3.8"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_NUM_PARTITIONS: 6
      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
      KAFKA_MIN_INSYNC_REPLICAS: 1
    healthcheck:
      test: kafka-broker-api-versions --bootstrap-server localhost:9092
      interval: 10s
      timeout: 10s
      retries: 5

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092

Lessons Learned

1. Always use acks=-1 for critical messages. The performance cost is minimal compared to the data loss risk with weaker acknowledgment settings.

2. Increase session.timeout.ms for consumers that process complex messages. The default 30 seconds causes premature rebalancing during processing spikes.

3. Implement dead letter queues for every topic that processes external input. Poison messages will eventually appear, and they must be handled gracefully.

4. Monitor consumer lag continuously. A growing lag indicates that consumers cannot keep up with producers, which requires scaling consumers or optimizing processing.

5. Use manual offset commits instead of auto-commit to ensure messages are only marked as processed after successful handling.

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.