[![NPM Version](https://img.shields.io/npm/v/kafka-pub-sub.svg?style=flat-square)](https://www.npmjs.com/package/kafka-pub-sub) [![NPM Monthly Downloads](https://img.shields.io/npm/dm/kafka-pub-sub?style=flat-square)](https://npmjs.org/package/kafka-pub-sub) [![Last Commit](https://img.shields.io/github/last-commit/mdmuhtasimfuadfahim/kafka-pub-sub?style=flat-square)](https://github.com/mdmuhtasimfuadfahim/kafka-pub-sub/commits/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg?style=flat-square)](LICENSE) # kafka-pub-sub Enterprise-grade Kafka publish/subscribe for Node.js, built on [KafkaJS](https://www.npmjs.com/package/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 ```sh npm install kafka-pub-sub ``` **Node.js ≥ 16 required.** You also need a running Kafka broker — see [Local Kafka with Docker](#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)** ```sh npm install dotenv ``` ```js // At the very top of your entry file, before requiring this package require('dotenv').config(); const { ProduceEvent, ConsumeEvent } = require('kafka-pub-sub'); ``` ```env # .env KAFKA_CLIENT_ID=my-service KAFKA_BROKER_URLS=localhost:9092 KAFKA_GROUP_ID=my-consumer-group ``` **Shell / CI** ```sh 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 ```sh 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. ```js 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: ```json { "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). ```js 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:** ```js { 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 `.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. ```js 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. ```js 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. ```js 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:** ```js 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. ```sh # 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: ```sh 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) ``` ```js // 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. ```js // 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. ```js // 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:** ```js 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:** ```env KAFKA_BROKER_URLS=pkc-xxxxx.us-east-1.aws.confluent.cloud:9092 KAFKA_SSL_ENABLED=true KAFKA_SASL_MECHANISM=plain KAFKA_SASL_USERNAME= KAFKA_SASL_PASSWORD= ``` **AWS MSK (IAM auth):** ```env 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 ```sh 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: ```js // 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](https://github.com/mdmuhtasimfuadfahim)