Event-Driven Node.js: Building Scalable Pub/Sub Messaging with Apache Kafka

Modern backend architectures demand systems that are resilient, decoupled, and capable of handling massive throughput in real time. When building microservices, traditional HTTP-based REST APIs often introduce tight coupling, cascading failures, and bottlenecks under heavy load.

Enter event-driven architecture (EDA) and Apache Kafka. By decoupling service communication through a distributed commit log, producers can publish events without knowing who the consumers are, and consumers can process messages at their own pace.

In this comprehensive guide, we will explore how to build a robust pub/sub messaging architecture in Node.js using KafkaJS, the modern, native Apache Kafka client for Node.js. We will move beyond basic “Hello World” examples and dive deep into production-grade patterns: consumer groups, partition management, manual offset committing, and at-least-once delivery semantics.

Why Apache Kafka and Node.js?

Node.js excels at I/O-bound, asynchronous tasks thanks to its event-driven, single-threaded event loop. However, native Node.js architectures can easily run into trouble if background processing or inter-service communication blocks the event loop.

Apache Kafka complements Node.js perfectly:

• High Throughput: Kafka can handle millions of messages per second with constant $O(1)$ disk reads and writes.
• Durability & Replayability: Unlike volatile message brokers (e.g., Redis Pub/Sub), Kafka persists messages to disk, allowing consumers to replay streams from any point in time.
• Partition Scaling: Kafka topics are partitioned across brokers, mapping naturally to horizontal scaling models in Node.js clusters.

Setting Up the Environment

Before writing code, ensure you have a running Kafka broker. The fastest way to spin up Kafka locally for development is via Docker Compose.

Create a docker-compose.yml file in your project root:

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

Run the cluster using:

docker-compose up -d

Installing Dependencies

Initialize a new Node.js project and install kafkajs:

mkdir kafka-node-app
cd kafka-node-app
npm init -y
npm install kafkajs dotenv

Configuring the Kafka Client

Let’s create a centralized configuration module (kafka.js) that initializes the Kafka client instance. This instance will be shared across our producers and consumers.

// kafka.js
const { Kafka, logLevel } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'order-processing-service',
  brokers: [process.env.KAFKA_BROKER || 'localhost:9092'],
  logLevel: logLevel.INFO,
  retry: {
    initialRetryTime: 100,
    retries: 8
  }
});

module.exports = kafka;

Building a Robust Kafka Producer

The producer is responsible for publishing events to a Kafka topic. In a resilient architecture, we want to ensure connection pooling, graceful shutdowns, and structured payloads.

Let’s build an order-producer.js script:

// order-producer.js
require('dotenv').config();
const kafka = require('./kafka');

const producer = kafka.producer({
  allowAutoTopicCreation: true,
  transactionalId: 'order-producer-tx-1'
});

const runProducer = async () => {
  try {
    await producer.connect();
    console.log('Kafka Producer connected successfully.');

    // Simulate producing order events
    const order = {
      orderId: 'ord_' + Math.floor(Math.random() * 100000),
      userId: 'usr_98765',
      items: [
        { productId: 'prod_abc', quantity: 2, price: 49.99 },
        { productId: 'prod_xyz', quantity: 1, price: 129.99 }
      ],
      totalAmount: 229.97,
      createdAt: new Date().toISOString()
    };

    const topic = 'order-events';
    
    // Send message with a partitioning key (ensures ordering per user/order)
    const result = await producer.send({
      topic,
      messages: [
        {
          key: order.userId,
          value: JSON.stringify(order),
          headers: { correlationId: 'corr_' + Date.now() }
        }
      ]
    });

    console.log(`Event published successfully to topic [${topic}]:`, JSON.stringify(result, null, 2));

  } catch (error) {
    console.error('Error publishing message:', error);
  } finally {
    await producer.disconnect();
  }
};

runProducer();

Key Producer Concept: Partition Keys

By supplying a key (order.userId), Kafka guarantees that all messages with the same key will be routed to the exact same partition. This preserves strict ordering semantics for per-user event streams.

Building an Advanced Consumer with Consumer Groups

Consumers read data from Kafka topics. To scale processing horizontally, Kafka uses Consumer Groups. When multiple consumer instances share the same groupId, Kafka automatically divides the partitions among them.

However, production apps require careful handling of offsets (pointers tracking which messages have been consumed) to prevent data loss.

Achieving At-Least-Once Delivery Semantics

By default, auto-committing offsets can lead to message loss if a Node.js process crashes after marking a message as read in memory, but before business logic successfully executes. To guarantee at-least-once delivery, we disable auto-commit and commit offsets manually only after successful processing.

Let’s build order-consumer.js:

// order-consumer.js
require('dotenv').config();
const kafka = require('./kafka');

const consumer = kafka.consumer({
  groupId: 'order-processing-group',
  // Disable auto-commit to take manual control of offsets
  sessionTimeout: 30000,
  heartbeatInterval: 3000
});

const processOrder = async (orderData) => {
  // Simulate business logic (database writes, external API calls, etc.)
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      if (Math.random() < 0.1) {
        return reject(new Error('Simulated Database Failure'));
      }
      console.log(`[Business Logic] Processed order: ${orderData.orderId}`);
      resolve();
    }, 500);
  });
};

const runConsumer = async () => {
  try {
    await consumer.connect();
    console.log('Kafka Consumer connected.');

    await consumer.subscribe({ topic: 'order-events', fromBeginning: true });

    await consumer.run({
      // Disable auto-commit
      autoCommit: false,
      eachMessage: async ({ topic, partition, message }) => {
        const rawValue = message.value.toString();
        const order = JSON.parse(rawValue);
        const correlationId = message.headers.correlationId?.toString();

        console.log(`Received message -> Partition: ${partition} | Offset: ${message.offset} | CorrelationID: ${correlationId}`);

        let retries = 3;
        let success = false;

        // Retry logic for transient failures
        while (retries > 0 && !success) {
          try {
            await processOrder(order);
            success = true;
          } catch (err) {
            retries--;
            console.warn(`Processing failed for order ${order.orderId}. Retries left: ${retries}. Error: ${err.message}`);
            if (retries === 0) {
              console.error(`CRITICAL: Moving order ${order.orderId} to Dead Letter Queue (DLQ).`);
              // TODO: Publish message to a DLQ topic here
            } else {
              // Backoff before retry
              await new Promise(res => setTimeout(res, 1000));
            }
          }
        }

        // Manually commit the offset ONLY after successful processing
        if (success) {
          try {
            await consumer.commitOffsets([
              {
                topic,
                partition,
                offset: (parseInt(message.offset, 10) + 1).toString(),
              },
            ]);
            console.log(`Offset ${message.offset} committed successfully for partition ${partition}.`);
          } catch (commitError) {
            console.error('Failed to commit offset:', commitError);
          }
        }
      },
    });
  } catch (error) {
    console.error('Consumer error:', error);
    await consumer.disconnect();
  }
};

runConsumer();

// Graceful shutdown handling
const shutdown = async () => {
  console.log('Disconnecting consumer gracefully...');
  await consumer.disconnect();
  process.exit(0);
};

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

Deep Dive: Partition Management & Rebalancing

When scaling out your Node.js backend by running multiple instances of order-consumer.js with the same groupId, Kafka triggers a rebalance. During a rebalance, partition assignments are recalculated across all active consumers.

Handling Rebalance Events

KafkaJS provides listeners to track consumer group rebalances, allowing you to flush state or save in-flight transaction offsets before partition ownership shifts:

const {льній, ConsumerEvents } = kafka;

consumer.on(consumer.events.GROUP_JOIN, event => {
  console.log('Consumer joined group:', event.payload);
});

consumer.on(consumer.events.REBALANCE, event => {
  console.log('Partition rebalance triggered:', event.payload);
});

If you prefer batch processing over single-message processing (e.g., bulk inserting logs or orders into PostgreSQL/MongoDB), you can swap eachMessage for eachBatch:

await consumer.run({
  autoCommit: false,
  eachBatch: async ({ batch, resolveOffset, heartbeat, commitOffsetsIfNecessary, isRunning }) => {
    const orders = [];
    for (let message of batch.messages) {
      if (!isRunning()) break;
      orders.push(JSON.parse(message.value.toString()));
      
      // Acknowledge individual message processing in batch
      resolveOffset(message.offset);
      await heartbeat();
    }

    // Perform bulk database operation
    await db.collection('orders').insertMany(orders);

    // Commit entire batch offsets
    await commitOffsetsIfNecessary();
    console.log(`Batch of ${orders.length} orders processed and committed.`);
  }
});

Error Handling and Dead Letter Queues (DLQ)

In distributed systems, poison messages (malformed payloads or persistent downstream failures) can cause consumer loops to stall infinitely. Implementing a Dead Letter Queue (DLQ) pattern isolates these problematic messages.

When a message exhausts its retry attempts inside your consumer loop, catch the exception, publish the payload to an order-events-dlq topic with diagnostic metadata (error stack, failure timestamp), and then commit the original offset so the consumer can keep moving forward.

Conclusion

Event-driven architectures powered by Apache Kafka and Node.js enable you to build highly scalable, resilient, and decoupled microservice ecosystems. By leveraging KafkaJS, you get granular control over:

1. Producer Keys: Guaranteeing strict ordering for related events.
2. Consumer Groups: Horizontally scaling processing capacity.
3. Manual Offset Committing: Achieving reliable at-least-once delivery semantics without risking data loss.

As you take these patterns into production, remember to monitor consumer lag closely, tune your partition counts based on throughput requirements, and implement robust DLQ strategies for unprocessable messages.