kafka-pub-sub

Enterprise-grade Kafka publish/subscribe for Node.js, built on KafkaJS.

Handles the operational complexity so your application code stays clean: persistent producer connections, automatic retry with exponential back-off, dead-letter queues, batch publishing, admin topic management, and health checks — all in one package.

Real-world use cases: e-commerce order pipelines, financial payment processing, GDPR/SOC 2 audit logging.

What's new in v2

• Persistent producer connection pool — no connect/disconnect per message
• Callback-based consumer — handler fires for every message, runs until you call stop()
• Dead-letter queue (DLQ) with automatic retry and exponential back-off
• Batch producer for high-throughput publishing across multiple topics
• SSL/TLS and SASL authentication (plain, SCRAM, OAuth)
• Multi-broker support
• Message compression (gzip, snappy, lz4, zstd)
• Idempotent producer (exactly-once semantics)
• Admin client for topic creation, listing, and consumer lag monitoring
• Health check suitable for Kubernetes liveness/readiness probes

Install

npm install kafka-pub-sub

Node.js ≥ 16 required. You also need a running Kafka broker — see Local Kafka with Docker below.

Configuration

This is a library — it reads process.env directly and never loads any .env file itself. Populating environment variables is your application's responsibility.

dotenv (local dev)

npm install dotenv

// At the very top of your entry file, before requiring this package
require('dotenv').config();
const { ProduceEvent, ConsumeEvent } = require('kafka-pub-sub');

# .env
KAFKA_CLIENT_ID=my-service
KAFKA_BROKER_URLS=localhost:9092
KAFKA_GROUP_ID=my-consumer-group

Shell / CI

export KAFKA_CLIENT_ID=my-service
export KAFKA_BROKER_URLS=localhost:9092
export KAFKA_GROUP_ID=my-consumer-group
node app.js

Platform-managed secrets (AWS ECS, Kubernetes, Heroku, etc.) — set the variables in your platform dashboard or manifest. No extra code needed.

Minimum required: KAFKA_CLIENT_ID, KAFKA_BROKER_URLS, KAFKA_GROUP_ID. All other variables have sensible defaults.

A copy of .env.example is included in the repository to document every available variable. It is not published to npm.

All configuration options

| Variable | Required | Default | Description | |---|---|---|---| | KAFKA_CLIENT_ID | ✅ | — | Service identifier shown in broker logs | | KAFKA_BROKER_URLS | ✅ | — | Comma-separated brokers: b1:9092,b2:9092 | | KAFKA_GROUP_ID | ✅ | — | Consumer group ID | | KAFKA_SSL_ENABLED | | false | Enable TLS | | KAFKA_SSL_CA_PATH | | — | Path to CA cert file | | KAFKA_SSL_KEY_PATH | | — | Path to client key file | | KAFKA_SSL_CERT_PATH | | — | Path to client cert file | | KAFKA_SASL_MECHANISM | | — | plain / scram-sha-256 / scram-sha-512 / oauthbearer | | KAFKA_SASL_USERNAME | | — | SASL username | | KAFKA_SASL_PASSWORD | | — | SASL password | | KAFKA_COMPRESSION_TYPE | | none | none / gzip / snappy / lz4 / zstd | | KAFKA_IDEMPOTENT | | false | Exactly-once producer semantics | | KAFKA_RETRY_INITIAL_TIME | | 300 | Initial retry delay in ms | | KAFKA_RETRY_COUNT | | 8 | Max broker connection retries | | KAFKA_SESSION_TIMEOUT | | 30000 | Consumer session timeout (ms) | | KAFKA_HEARTBEAT_INTERVAL | | 3000 | Consumer heartbeat interval (ms) | | KAFKA_MAX_WAIT_TIME | | 5000 | Max time broker waits before returning empty fetch (ms) | | KAFKA_LOG_LEVEL | | error | error / warn / info / debug |

Local Kafka with Docker

docker compose up -d
# Kafka 3.8 (KRaft, no ZooKeeper) on localhost:9092
# Kafka UI at http://localhost:8080

Important: Always create topics before starting a consumer. Subscribing to a topic that does not exist yet will fail. Use KafkaAdmin (see below) or the Kafka UI to pre-create topics.

API Reference

ProduceEvent(topic, event, data?, headers?, options?)

Publishes a single message. The producer connection is created once and reused across all calls.

const { ProduceEvent } = require('kafka-pub-sub');

await ProduceEvent(
  'order.placed',                              // topic
  'ORDER_PLACED',                              // event name (becomes message key prefix)
  { orderId: 'ORD-001', total: 149.99 },       // payload
  { 'x-source': 'checkout-service' },          // custom headers (optional)
  {
    partitionKey: 'CUST-42',    // routes all messages for this customer to same partition
    correlationId: 'req-xyz',   // automatically added as correlation-id header
    compression: 'gzip',        // per-message compression override
  }
);

// Call once during graceful shutdown:
await ProduceEvent.disconnect();

Topic naming: [a-zA-Z0-9._-], max 249 characters (Kafka's hard limit). Use dot-separated namespaces: payment.transaction, audit.user-action.

The message value written to Kafka is:

{ "event": "ORDER_PLACED", "data": { "orderId": "ORD-001", "total": 149.99 }, "timestamp": 1700000000000 }

ConsumeEvent(topic, handler, options?)

Subscribes to a topic. Calls handler(message) for every incoming message. Returns a stop() function for graceful shutdown.

The topic must already exist before calling ConsumeEvent. Create it with KafkaAdmin.createTopics first (see below).

const { ConsumeEvent } = require('kafka-pub-sub');

const stop = await ConsumeEvent(
  'order.placed',
  async (msg) => {
    console.log(msg.value.event);    // 'ORDER_PLACED'
    console.log(msg.value.data);     // { orderId: 'ORD-001', total: 149.99 }
    console.log(msg.headers);        // decoded headers object
    console.log(msg.offset);         // Kafka offset string
  },
  {
    fromBeginning: false,  // consume from latest (default); true = replay from offset 0
    retry: 3,              // handler retries with exponential back-off before DLQ
    dlq: true,             // failed messages → order.placed.dlq (default: true)
    groupId: 'my-group',   // override KAFKA_GROUP_ID for this consumer
  }
);

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

Message object received by handler:

{
  topic:     'order.placed',
  partition: 0,
  offset:    '42',
  timestamp: '1700000000000',
  key:       'key-ORDER_PLACED',
  value:     { event: 'ORDER_PLACED', data: { ... }, timestamp: 1700000000000 },
  headers:   { 'correlation-id': '...', 'x-source': '...' }
}

DLQ behaviour: when dlq: true (default), a message that fails all retries is automatically forwarded to <topic>.dlq with headers dlq-original-topic, dlq-failed-at, and dlq-error. Run a separate consumer on the DLQ topic to inspect or replay failed messages.

Retry timing: 100 ms → 200 ms → 400 ms → … (exponential back-off).

BatchProduceEvent(messages, options?)

Sends multiple messages across one or more topics in a single broker round-trip. More efficient than calling ProduceEvent in a loop.

const { BatchProduceEvent } = require('kafka-pub-sub');

await BatchProduceEvent(
  [
    { topic: 'order.placed',    event: 'ORDER_PLACED',    data: { orderId: '1' } },
    { topic: 'inventory.check', event: 'INVENTORY_CHECK', data: { sku: 'SKU-99' } },
    { topic: 'order.placed',    event: 'ORDER_PLACED',    data: { orderId: '2' } },
  ],
  {
    compression: 'gzip',
    correlationId: 'batch-001',  // applied to every message in the batch
  }
);

await BatchProduceEvent.disconnect();

Messages are automatically grouped by topic before being sent, so you can mix topics freely in the array.

KafkaAdmin()

Returns an admin client for topic management. Always call disconnect() when finished.

const { KafkaAdmin } = require('kafka-pub-sub');

const admin = KafkaAdmin();

// Create topics (safe to call even if topics already exist)
await admin.createTopics([
  {
    topic: 'payment.transaction',
    numPartitions: 24,
    replicationFactor: 3,
    configEntries: { 'retention.ms': '604800000' },  // 7-day retention
  },
]);

// List all topics (filters out internal __ topics)
const topics = await admin.listTopics();

// Check whether specific topics exist
const { exists, missing } = await admin.topicsExist(['order.placed', 'does-not-exist']);
// exists → false (not ALL listed topics exist)
// missing → ['does-not-exist']

// Consumer lag monitoring
const offsets = await admin.getConsumerGroupOffsets('payment-processor', 'payment.transaction');

await admin.disconnect();

HealthCheck(options?)

Probes the broker and returns a status object. Use as a Kubernetes liveness/readiness probe.

const { HealthCheck } = require('kafka-pub-sub');

const status = await HealthCheck({ timeout: 5000 });
// Healthy:
// { healthy: true, brokerCount: 1, brokers: ['localhost:9092'], clusterId: 'abc123', checkedAt: '...' }
// Unhealthy:
// { healthy: false, error: 'Connection timeout', checkedAt: '...' }

Express.js probe:

app.get('/health/kafka', async (req, res) => {
  const status = await HealthCheck();
  res.status(status.healthy ? 200 : 503).json(status);
});

Local testing workflow

These steps use the bundled kafka-pub-sub-test/ project which is a standalone app that depends on this package.

# 1. Start Kafka
docker compose up -d
# Wait ~15 seconds, then check http://localhost:8080

# 2. Register the package globally via npm link
npm install
npm link

# 3. Set up the test app
cd kafka-pub-sub-test
npm install
npm link kafka-pub-sub

# 4. Pre-create topics (required before subscribing)
node admin.js

# 5. Terminal 1 — start the subscriber
node subscribe.js

# 6. Terminal 2 — publish messages
node publish.js       # sends 3 messages
node publish.js 10    # sends 10 messages

Why pre-create topics? Kafka requires that a topic's partition metadata be available before a consumer can be assigned to it. node admin.js creates test.events and test.events.dlq once. After that, you can start and stop the subscriber freely.

Verify the symlink is pointing to your local package:

ls -la node_modules/kafka-pub-sub
# → /Users/you/kafka-pub-sub  (symlink)

Industry examples

E-commerce order pipeline

A saga pattern where each step in the order lifecycle emits an event that triggers the next service. Services are decoupled — they only know about the topic they consume and produce.

checkout-service  → order.placed
                       ↓
inventory-service → inventory.reserved  (or inventory.failed → order.cancelled)
                       ↓
payment-service   → payment.charged     (or payment.failed → order.cancelled)
                       ↓
fulfillment-service → fulfillment.shipped
                       ↓
notification-service → (email/SMS to customer)

// checkout-service: emit when customer places order
await ProduceEvent('order.placed', 'ORDER_PLACED', {
  orderId: 'ORD-9001',
  customerId: 'CUST-42',
  items: [{ sku: 'SHOE-RED-10', qty: 1, price: 89.99 }],
  total: 89.99,
  currency: 'USD',
}, {}, { partitionKey: 'CUST-42', correlationId: 'sess-abc' });

// inventory-service: consume and reserve stock
const stop = await ConsumeEvent('order.placed', async (msg) => {
  const { orderId, items } = msg.value.data;
  await reserveStock(items);
  await ProduceEvent('inventory.reserved', 'INVENTORY_RESERVED', { orderId });
}, { groupId: 'inventory-service', retry: 5, dlq: true });

When to use this pattern: order management, multi-step checkout flows, any workflow where services need to react to each other without direct HTTP coupling.

Financial payments with DLQ and fraud detection

High-value transactions require exactly-once delivery and immediate routing to a fraud review queue when something looks wrong. The idempotent producer prevents duplicate charges if the broker retries a produce request.

// payment-service: produce with idempotent guarantee
// Set KAFKA_IDEMPOTENT=true in env
await ProduceEvent(
  'payment.transaction',
  'PAYMENT_INITIATED',
  {
    transactionId: 'TXN-8821',
    amount: 15000.00,
    currency: 'USD',
    fromAccount: 'ACC-001',
    toAccount:   'ACC-002',
    method: 'wire',
  },
  { 'x-idempotency-key': 'TXN-8821' },
  { partitionKey: 'ACC-001' }   // all transactions for same account → same partition → ordered
);

// fraud-detection-service: consume and score
const stop = await ConsumeEvent('payment.transaction', async (msg) => {
  const score = await runFraudModel(msg.value.data);
  if (score > 0.85) {
    await ProduceEvent('payment.fraud-alert', 'FRAUD_ALERT', {
      transactionId: msg.value.data.transactionId,
      score,
      reason: 'high-velocity',
    });
  }
}, { groupId: 'fraud-detection', retry: 3, dlq: true });

// ops team: monitor the DLQ for transactions that failed fraud scoring
const stopDlq = await ConsumeEvent('payment.transaction.dlq', async (msg) => {
  await alertOpsTeam(msg);
}, { groupId: 'payment-dlq-monitor', fromBeginning: true });

When to use this pattern: payment processing, banking transfers, any financial event that must not be duplicated or lost.

GDPR / SOC 2 audit logging

Every user action, data access, and config change is written to an append-only audit topic. Retention is set to 7 years. Consumers build compliance dashboards, trigger DSAR workflows, and feed a SIEM.

// Any service: emit an audit event
await ProduceEvent(
  'audit.user-action',
  'USER_LOGIN',
  {
    userId:     'USR-991',
    action:     'LOGIN',
    resource:   '/dashboard',
    ipAddress:  '203.0.113.5',
    userAgent:  'Mozilla/5.0 ...',
    result:     'SUCCESS',
    timestamp:  new Date().toISOString(),
  },
  { 'x-service': 'auth-service', 'x-region': 'eu-west-1' }
);

// compliance-service: build immutable audit trail
const stop = await ConsumeEvent('audit.user-action', async (msg) => {
  await writeToAuditDatabase(msg.value.data);
  await updateComplianceDashboard(msg.value.data);
}, { groupId: 'compliance-service', fromBeginning: true, retry: 5, dlq: true });

Topic setup for 7-year retention:

const admin = KafkaAdmin();
await admin.createTopics([
  {
    topic: 'audit.user-action',
    numPartitions: 12,
    replicationFactor: 3,
    configEntries: {
      'retention.ms':     String(7 * 365 * 24 * 60 * 60 * 1000),  // 7 years
      'cleanup.policy':   'delete',
      'min.insync.replicas': '2',
    },
  },
]);
await admin.disconnect();

When to use this pattern: GDPR compliance, SOC 2 audit trails, HIPAA access logging, security incident investigation.

Production: cloud brokers

Confluent Cloud:

KAFKA_BROKER_URLS=pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
KAFKA_SSL_ENABLED=true
KAFKA_SASL_MECHANISM=plain
KAFKA_SASL_USERNAME=<API_KEY>
KAFKA_SASL_PASSWORD=<API_SECRET>

AWS MSK (IAM auth):

KAFKA_BROKER_URLS=b-1.xxx.kafka.us-east-1.amazonaws.com:9098,b-2.xxx.kafka.us-east-1.amazonaws.com:9098
KAFKA_SSL_ENABLED=true
KAFKA_SASL_MECHANISM=oauthbearer

Testing

npm test           # unit tests — no broker required, kafkajs is fully mocked
npm run coverage   # coverage report

Migration from v1

The consumer API is a breaking change:

// v1 — resolved once on the first message only
const data = await ConsumeEvent('MY_TOPIC');

// v2 — callback invoked for every message; returns stop()
const stop = await ConsumeEvent('MY_TOPIC', async (msg) => {
  console.log(msg.value);
});
await stop();

The env variable KAFKA_BROKER_URL (singular) is renamed to KAFKA_BROKER_URLS (plural, comma-separated for multi-broker).

License

MIT © Md. Muhtasim Fuad Fahim