Build Consumer

const { Kafka } = require('kafkajs')

const kafka = new Kafka({
  // User-specific properties that you must set
  clientId: 'my-app',
  brokers: ['<BOOTSTRAP SERVERS>'],

  // Fixed properties
  ssl: true,
  sasl: {
    mechanism: 'plain',
    username: 'unused',
    password: 'token:<API KEY>',
  },
})

async function receive() {
  let topic = 'purchases'

  const consumer = kafka.consumer({
    groupId: 'kafka-nodejs-getting-started',
  })
  await consumer.connect()
  console.log('Connected to Kafka')

  await consumer.subscribe({ topic: topic, fromBeginning: true })

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      let k = message.key.toString().padEnd(10, ' ')
      let value = message.value.toString()
      console.log(
        `Consumed event from topic ${topic}: key = ${k} value = ${value}`
      )
    },
  })
}

receive().catch((err) => {
  console.error(`Something went wrong:\n${err}`)
  process.exit(1)
})