Node.js Kafka Producer

Overview

This application demonstrates a robust logging architecture using modern Node.js libraries. The flow: Node.js App → Kafka Topic → Logstash → Elasticsearch

Architecture Features

winston — logging framework with multiple transport support
kafkajs — modern, robust Kafka client for Node.js
Custom Winston transport for seamless Kafka integration
Structured JSON logs with metadata per record
Generates random logs every second for testing

Configuration Variables

These are the ONLY variables you need to modify according to your environment:

Variable	Default Value	Description
`KAFKA_BROKER_HOST`	`'192.168.20.208:9092'`	REQUIRED: Replace with your Kafka broker IP and port
`KAFKA_TOPIC`	`'application_logs'`	Topic name — must match the topic created in Kafka
`APPLICATION_NAME`	`'node-kafka-log-producer'`	Your application identifier for log tracking

⚠ Important: Replace KAFKA_BROKER_HOST with your actual Kafka broker address before running!

01 Install Dependencies

bash

npm install winston kafkajs winston-transport

Create Application File

Save the code as app.js or kafka-producer.js, update the config variables, then run:

bash

node app.js

02 Complete Application Code

Copy this entire code into your Node.js application file:

javascript

/**
 * Kafka Log Producer Application (Node.js)
 * Dependencies: winston, kafkajs, winston-transport
 * Install: npm install winston kafkajs winston-transport
 * Flow: Node.js App -> Kafka Topic -> Logstash -> Elasticsearch
 */
const winston = require('winston');
const Transport = require('winston-transport');
const { Kafka } = require('kafkajs');
const { combine, timestamp, json } = winston.format;

// --- Configuration --- MODIFY THESE ---
const KAFKA_BROKER_HOST = '192.168.20.208:9092';
const KAFKA_TOPIC = 'application_logs';
const APPLICATION_NAME = 'node-kafka-log-producer';
// --------------------------------------

const logLevels = ['info', 'warn', 'error'];
const logMessages = [
    'User logged in successfully.',
    'Database connection timeout.',
    'Processing order: #12345',
    'Warning: API rate limit approaching.',
    'Error: File not found in path /app/data.',
    'Service health check passed.'
];

// --- Custom KafkaJs Transport Class ---
class KafkaJsTransport extends Transport {
    constructor(opts) {
        super(opts);
        this.topic = opts.topic;
        this.kafka = new Kafka({
            clientId: APPLICATION_NAME,
            brokers: [opts.brokerList]
        });
        this.producer = this.kafka.producer();
        this.producer.connect()
            .then(() => console.log(`Kafka Producer connected to ${opts.brokerList}`))
            .catch(e => {
                console.error(`!!! KAFKA CONNECTION ERROR !!!`);
                console.error(`Could not connect to broker: ${opts.brokerList}`, e.message);
            });
    }

    log(info, callback) {
        setImmediate(() => { this.emit('logged', info); });
        const logData = {
            ...info.data,
            ...info.metadata,
            timestamp: info.timestamp,
            level: info.level,
            message: info.message,
        };
        this.producer.send({
            topic: this.topic,
            messages: [{ value: JSON.stringify(logData) }],
        })
        .catch((e) => {
            console.error(`!!! KAFKA SEND ERROR !!! ${e.message}`);
        })
        .finally(() => { callback(); });
    }
}

// Create Kafka Transport
const kafkaTransport = new KafkaJsTransport({
    brokerList: KAFKA_BROKER_HOST,
    topic: KAFKA_TOPIC,
    metadata: {
        application: APPLICATION_NAME,
        environment: 'production',
    }
});

// Create Winston Logger
const logger = winston.createLogger({
    level: 'info',
    format: combine(
        timestamp({ format: 'YYYY-MM-DD HH:mm:ss.SSS' }),
        json()
    ),
    transports: [
        new winston.transports.Console({ format: winston.format.simple() }),
        kafkaTransport
    ]
});

// Log Generator
function generateRandomLog() {
    const level = logLevels[Math.floor(Math.random() * logLevels.length)];
    const message = logMessages[Math.floor(Math.random() * logMessages.length)];
    logger.log({
        level,
        message,
        data: {
            processId: process.pid,
            requestId: Math.random().toString(36).substring(2, 9),
            logSource: 'application-backend'
        }
    });
    console.log(`[SENT] Level: ${level}, Message: "${message}"`);
}

console.log(`Starting Kafka Producer...`);
console.log(`Target Broker: ${KAFKA_BROKER_HOST}`);
console.log(`Target Topic: ${KAFKA_TOPIC}`);
setInterval(generateRandomLog, 1000);

Code Explanation

Custom Transport Class

Component	Purpose
`constructor()`	Initializes Kafka client and connects the producer to the broker
`log(info, callback)`	Formats log data as JSON and sends it to the Kafka topic
`producer.send()`	Asynchronously publishes the message to Kafka

Log Generation

Each generated log includes:

Random log level — info, warn, or error
Random message from predefined list
Dynamic metadata: processId, requestId, logSource
Automatic timestamp from Winston

Expected Output

console output

Starting Kafka Producer...
Target Broker: 192.168.20.208:9092
Target Topic: application_logs
Kafka Producer connected to 192.168.20.208:9092
[SENT] Level: info, Message: "User logged in successfully."
[SENT] Level: warn, Message: "Warning: API rate limit approaching."
[SENT] Level: error, Message: "Error: File not found in path /app/data."
[SENT] Level: info, Message: "Service health check passed."

Troubleshooting

Connection Error

Error: !!! KAFKA CONNECTION ERROR !!!

Solution: Verify KAFKA_BROKER_HOST is the correct IP and port. Ensure Kafka is running and accessible from your network.

Send Error

Error: !!! KAFKA SEND ERROR !!!

Solution: Check the topic exists in Kafka and the producer has write permission. Verify the topic name matches exactly.

Module Not Found

Error: Cannot find module 'winston'

Solution: Run npm install winston kafkajs winston-transport