Example for fastify-kafka

[vincentbitter]

💬 Question here

I tried to setup fastify-kafka, but the documentation seems outdated. Finally I managed to produce messages, but consuming those is a big mystery. In the documentation (README) it shows the following example code:

fastify.kafka.subscribe('updates')
fastify.kafka.on('updates', (msg, commit) => {
  console.log(msg.value.toString())
  commit()
})

However, the on methos is not available on the kafka object. It is on the return object of subscribe(), but that one only allows me to use things like'data' and not 'updates'. I am using Typescript by the way.

Does anyone have a good example of how to implement fastify-kafka with the plugins and routes folders?

Your Environment

• node version: 18
• fastify version: >=4.13.0
• @fastify/kafka: ">=2.2.1",
• os: docker.io/node:lts-alpine

[Eomm]

Could you add a Minimal, Reproducible Example?

Without it, we are unable to help you.

Otherwise, you may take a look in the test folder:
https://github.com/fastify/fastify-kafka/blob/master/test/communication.test.js

[enzolupia]

HI @Eomm same issue here.
It looks like there is something strange with the example in the README on https://github.com/fastify/fastify-kafka .
Same for the examples in the /examples folder.

@fastify/kafka verion: 2.2.1

'use strict'

const crypto = require('node:crypto')
const fastify = require('fastify')({
  logger: {
    level: 'debug'
  }
})

const groupId = crypto.randomBytes(20).toString('hex')

fastify
  .register(require('..'), {
    consumer: {
      'metadata.broker.list': '127.0.0.1:9092',
      'group.id': groupId,
      'fetch.wait.max.ms': 10,
      'fetch.error.backoff.ms': 50
    },
    consumerTopicConf: {
      'auto.offset.reset': 'earliest'
    }
  })
  .after(err => {
    if (err) throw err

    fastify.kafka.consumer.on('error', err => { if (err) throw err })

    fastify.kafka.subscribe(['test'])

    fastify.kafka.on('test', (msg, commit) => {
      console.log(msg.value.toString())
      commit()
    })

    fastify.kafka.consume()

  })

Please note that neither fastify.kafka.consumer nor fastify.kafka export an 'on' method, thus resulting in
TS2339: Property  on  does not exist on type  FastifyKafkaConsumer  and TS2339: Property  on  does not exist on type  Kafka  errors.

@vincentbitter did you manage to find a working configuration in the meanwhile?

[vincentbitter]

I was able to get an example running at that moment in time. It was in a project created with NX and running on TypeScript, so I had to create a new example for this purpose.

@Eomm Reproduction now resulted in a different error: https://github.com/vincentbitter/fastify-kafka-issue

@enzolupia Please check the pull requests at my repo: https://github.com/vincentbitter/fastify-kafka-issue/pulls
Fix 1: simple solution
Fix 2: read in bulk