Kafka consume message

gistfile1.txt

//Sample kafka client in nodejs

'use strict';
require('dotenv').config()

//kafka init starts

const { Kafka } = require('kafkajs')
const kafka = new Kafka({
	clientId: process.env.consumerid, 
	brokers: [ process.env.bootstrapservers ],	
	ssl: true,
	sasl: {
		mechanism: process.env.saslmechanisms,
		username:  process.env.saslusername,
		password:  process.env.saslpassword
	},
})

//kafka init ends


//consumer starts

const consumer = kafka.consumer({ groupId: process.env.groupid }) 
consumer.connect()
consumer.subscribe({
	topic: process.env.topic, fromBeginning: true
})

consumer.run({
	eachMessage: async ({ topic, partition, message }) => {
		console.log({
			value: message.value.toString(),
		})
	},
})

//consumer end