Apache Kafka with Node.js: Event Streaming for Microservices

Kafka Core Concepts

• Topic: Named stream of records
• Partition: Ordered, immutable sequence within a topic
• Offset: Position of a record in a partition
• Consumer Group: Parallel consumers sharing partitions

Setup with KafkaJS

npm install kafkajs

import { Kafka, Partitioners } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka1:9092', 'kafka2:9092'],
  ssl: true,
  sasl: {
    mechanism: 'scram-sha-256',
    username: process.env.KAFKA_USERNAME,
    password: process.env.KAFKA_PASSWORD,
  },
  retry: {
    initialRetryTime: 100,
    retries: 8,
  },
});

Producer

const producer = kafka.producer({
  createPartitioner: Partitioners.LegacyPartitioner,
  idempotent: true,           // Exactly-once delivery
  transactionTimeout: 30000,
});

await producer.connect();

// Send messages
await producer.send({
  topic: 'user-events',
  messages: [
    {
      key: userId,             // Ensures ordering per user
      value: JSON.stringify({
        type: 'USER_CREATED',
        userId,
        email,
        timestamp: Date.now(),
      }),
      headers: {
        'correlation-id': requestId,
        'service': 'user-service',
      },
    },
  ],
});

// Transactional producer (exactly-once)
const transaction = await producer.transaction();
try {
  await transaction.send({
    topic: 'orders',
    messages: [{ key: orderId, value: JSON.stringify(order) }],
  });
  await transaction.sendOffsets({
    consumerGroupId: 'order-processor',
    topics: [{ topic: 'payments', partitions: [{ partition: 0, offset: '42' }] }],
  });
  await transaction.commit();
} catch (err) {
  await transaction.abort();
  throw err;
}

Consumer with Graceful Shutdown

const consumer = kafka.consumer({ groupId: 'user-processor' });

await consumer.connect();
await consumer.subscribe({ topics: ['user-events'], fromBeginning: false });

await consumer.run({
  eachBatchAutoResolve: false,
  eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning }) => {
    for (const message of batch.messages) {
      if (!isRunning()) break;
      
      try {
        const event = JSON.parse(message.value.toString());
        await processEvent(event);
        resolveOffset(message.offset);
        await heartbeat(); // Prevent session timeout for slow processing
      } catch (err) {
        // Send to dead letter queue
        await dlqProducer.send({
          topic: 'user-events-dlq',
          messages: [{
            key: message.key,
            value: message.value,
            headers: {
              ...message.headers,
              'error': err.message,
              'original-topic': 'user-events',
            },
          }],
        });
        resolveOffset(message.offset); // Skip bad message
      }
    }
  },
});

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

Dead Letter Queue Pattern

// DLQ consumer for manual inspection/retry
const dlqConsumer = kafka.consumer({ groupId: 'dlq-processor' });
await dlqConsumer.subscribe({ topic: 'user-events-dlq' });

await dlqConsumer.run({
  eachMessage: async ({ message }) => {
    const headers = message.headers;
    const errorMsg = headers.error?.toString();
    const originalTopic = headers['original-topic']?.toString();
    
    // Log to monitoring system
    await alerting.send({
      title: 'Message processing failed',
      topic: originalTopic,
      error: errorMsg,
      payload: message.value.toString(),
    });
    
    // Optionally retry after N hours
    const retryCount = parseInt(headers['retry-count'] ?? '0');
    if (retryCount < 3) {
      await producer.send({
        topic: originalTopic,
        messages: [{
          ...message,
          headers: {
            ...headers,
            'retry-count': String(retryCount + 1),
          },
        }],
      });
    }
  },
});

Schema Registry with Avro

import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';

const registry = new SchemaRegistry({ host: 'http://schema-registry:8081' });

// Encode message
const schemaId = await registry.getLatestSchemaId('user-events-value');
const encodedValue = await registry.encode(schemaId, {
  type: 'USER_CREATED',
  userId: '123',
  email: 'user@example.com',
});

// Decode message in consumer
const decodedValue = await registry.decode(message.value);

Kafka Best Practices

• Partition strategy: Use message key for ordering guarantees
• Replication factor: min 3 in production
• Retention: Set based on consumer SLAs (default 7 days)
• Monitoring: Track consumer lag — alert if > threshold
• Idempotent consumers: Design for at-least-once delivery