Event Streaming with Apache Kafka and Node.js

Build resilient, growth-ready event-driven systems using Apache Kafka. Here is how to producer configuration, consumer groups with auto-rebalancing, offset management, and exactly-once processing semantics for reliable asynchronous communication between microservices.

When to Use This

• Services need to communicate asynchronously without tight coupling. See Event-Driven Microservices for architecture patterns.
• Event history must be replayable for debugging or new consumer onboarding. See Event Sourcing for immutable event logs.
• High throughput message processing requires horizontal scaling of consumers. See RabbitMQ Task Queue for alternative broker patterns.

Solution

1. Kafka Producer

// kafka/producer.ts
import { Kafka, Partitioners } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['kafka-1:9092', 'kafka-2:9092'],
});

const producer = kafka.producer({
  createPartitioner: Partitioners.DefaultPartitioner,
  retry: {
    retries: 5,
    initialRetryTime: 300,
  },
});

await producer.connect();

async function publishOrderCreated(order: unknown): Promise<void> {
  await producer.send({
    topic: 'orders.created',
    messages: [
      {
        key: order.userId,
        value: JSON.stringify(order),
        headers: {
          'content-type': 'application/json',
          'trace-id': generateTraceId(),
        },
      },
    ],
  });
}

2. Consumer with Consumer Group

// kafka/consumer.ts
const consumer = kafka.consumer({
  groupId: 'notification-service',
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
});

await consumer.connect();
await consumer.subscribe({ topic: 'orders.created', fromBeginning: false });

await consumer.run({
  autoCommit: true,
  autoCommitInterval: 5000,
  eachMessage: async ({ topic, partition, message }) => {
    const order = JSON.parse(message.value!.toString());
    console.log(`Processing order from partition ${partition}:`, order.id);

    try {
      await sendEmailNotification(order);
    } catch (error) {
      // Dead letter handling
      await publishToDeadLetter(topic, message, error);
    }
  },
});

3. Exactly-Once Processing

// kafka/exactly-once.ts
const producer = kafka.producer({
  transactionalId: 'order-processor',
  maxInFlightRequests: 1,
  idempotent: true,
});

await producer.connect();

async function processOrderWithIdempotency(orderId: string): Promise<void> {
  const transaction = await producer.transaction();

  try {
    // Process order
    const result = await processPayment(orderId);

    // Send result
    await transaction.send({
      topic: 'orders.completed',
      messages: [{ key: orderId, value: JSON.stringify(result) }],
    });

    // Commit offsets and messages atomically
    await transaction.commit();
  } catch (error) {
    await transaction.abort();
    throw error;
  }
}

4. Custom Partitioner for Ordering

// kafka/partitioner.ts
function userIdPartitioner(userId: string, numPartitions: number): number {
  // Ensure all events for a user go to the same partition
  let hash = 0;
  for (let i = 0; i < userId.length; i++) {
    hash = ((hash << 5) - hash) + userId.charCodeAt(i);
    hash |= 0;
  }
  return Math.abs(hash) % numPartitions;
}

await producer.send({
  topic: 'user.events',
  messages: [
    {
      key: userId,
      value: JSON.stringify(event),
      partition: userIdPartitioner(userId, 12),
    },
  ],
});

5. Docker Compose Setup

# docker-compose.kafka.yml
version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 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://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

6. Python Consumer with kafka-python

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'orders.created',
    bootstrap_servers=['localhost:9092'],
    group_id='notification-service',
    auto_offset_reset='latest',
    enable_auto_commit=False,
    value_deserializer=lambda x: json.loads(x.decode('utf-8')),
)

for message in consumer:
    order = message.value
    try:
        send_email_notification(order)
        consumer.commit()
    except Exception as e:
        print(f"Failed to process order {order['id']}: {e}")
        # Do not commit — message will be reprocessed on restart

Manual commit gives you control over when offsets are saved. Only commit after successful processing to avoid losing messages on failure.

How It Works

• Producers publish messages to topics partitioned across brokers
• Consumer groups distribute partitions among instances for parallel processing
• Offsets track consumer progress; auto-commit periodically persists position
• Exactly-once uses transactions to commit offsets and output messages atomically

Production Considerations

• Run at least 3 Kafka brokers with replication factor 3 for fault tolerance
• Monitor consumer lag with tools like Kafka Lag Exporter
• Use schema registry (Confluent) to enforce Avro/Protobuf schemas on topics
• Set appropriate retention policies — time-based (7 days default) or size-based
• Enable log compaction for topics that store current state (e.g., user profiles) rather than events
• Use ack=all producer setting to ensure messages are written to all replicas before confirming

Best Practices

• Use meaningful topic names: orders. created not topic1. Namespace by domain and event type.
• Partition by key: use a meaningful key (e. g. , userId, orderId) to ensure ordering within a partition.
• Batch producer sends: sending messages in batches improves throughput. sizeandlinger. ms`.
• Handle rebalances gracefully: implement a rebalance listener to commit offsets and clean up resources before partitions are revoked.
• Use idempotent producers: set enable. idempotence=true to prevent duplicate messages from retries.

Common Mistakes

• Not handling consumer rebalances, causing duplicate processing
• Using auto-commit with long-running processes that may fail mid-batch
• Creating too many partitions per topic, increasing coordination overhead
• Not setting a transactionalId when using transactions, causing producer fencing errors
• Ignoring consumer lag until it becomes critical — set alerts at 1000+ messages behind
• Using default partitioner when message ordering matters — use key-based partitioning instead
• Not configuring max.poll.interval.ms correctly — consumers that process slowly get kicked out of the group
• Using auto_offset_reset=none without committed offsets — consumers crash on first run

Troubleshooting

• Messages are lost on restart: persist messages before acknowledging.
• Consumer lags behind producer: scale consumers, increase prefetch, and partition the topic.
• Duplicate messages: design consumers to be idempotent.
• Ordering is wrong after scaling: preserve partition keys and avoid rebalancing during bursts. Consider a single partition when order is mandatory.
• Queue depth grows but consumers are idle: check network partitions, consumer health, and permission issues. Restart gracefully.

Further Reading

• Official documentation: check the current reference for the framework or tool used.
• Related guides: explore the event-driven and messaging guides for deeper coverage.
• Complementary patterns: review design patterns applicable to your technology stack.
• Public postmortems: study real incidents from teams that faced similar production issues.

Production Notes

• Deploy gradually using canary or blue-green to catch regressions early.
• Configure alerts for error rate, p99 latency, and failure rate before enabling in production.
• Document the rollback in the runbook; test the procedure in staging at least once per quarter.
• Review structured logs with correlation IDs to trace requests end-to-end during incidents.

Key Takeaways

• Apply event streaming with apache kafka and node.js when you need a practical solution for your use case.
• Monitor performance after implementation; measure latency, errors, and resource usage before and after.
• Check the Troubleshooting section for common failures; most have documented root causes with fixes.
• Keep dependencies updated and run tests in CI to prevent production regressions.

Common Production Pitfalls

• Copying the example without adapting it to real data volumes and failure modes.
• Skipping load and error-injection tests before the first production deployment.
• Hard-coding values that should be configurable per environment.
• Forgetting to add logging and monitoring at each step.
• Deploying without a rollback plan or a tested backup strategy.
• Assuming the minimal example will scale without adding caching or batching.
• Not documenting the version and configuration used in production.
• Letting the recipe sit unchanged when dependencies or scale evolve.