kafkajs_quickstart_produce_and_consume_message.ts

1import { Kafka, Message, Producer, Consumer, EachMessagePayload } from 'kafkajs'
2
3const kafka = new Kafka({
4  clientId: 'my-app',
5  brokers: ['localhost:9092'],
6})
7
8const producer: Producer = kafka.producer()
9const consumer: Consumer = kafka.consumer({ groupId: 'test-group' })
10
11const run = async (): Promise<void> => {
12  // Producing
13  await producer.connect()
14  await producer.send({
15    topic: 'test-topic',
16    messages: [
17      { value: 'Hello KafkaJS user!' },
18    ],
19  })
20
21  // Consuming
22  await consumer.connect()
23  await consumer.subscribe({ topic: 'test-topic', fromBeginning: true })
24
25  await consumer.run({
26    eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {
27      console.log({
28        partition,
29        offset: message.offset,
30        value: message.value?.toString(),
31      })
32    },
33  })
34}
35
36run().catch(console.error)