kafkajs run eachMessage is not fetching messages

Gil Allen 0
Viewed 38

I'm trying to fetch messages from Kafka using kafkajs I want to catch all messages into "messages" (an array) config - the variable which hold the configs Here is the function I'm using:

const { Kafka, logLevel } = require('kafkajs')
async function consume_messages(config) {
    const kafka = new Kafka({
        logLevel: logLevel.INFO,
        brokers: [config.broker],
        ssl: true,
        sasl: {
            mechanism: [config.mechanism], 
            username: config.Username,
            password: config.Password
        },
    })
    const topic = config.client_id
    const consumer = kafka.consumer({
        groupId: 'my-group', fromBeginning: true
    })
    await consumer.connect();
    await consumer.subscribe({
        topics: [topic],
        fromBeginning: true
    })
    let messages = []
    await consumer.run({
        eachMessage: async ({ message }) => {
             messages.push(message)
             console.log('RECEIVED MESSAGE', JSON.parse(message.value), message.offset);
          }
        })
       
    })
    await consumer.disconnect();
    return messages ;
}

when I run it I get nothing the "messages" stays empty However, when I removed the await from "await consumer.run({" -> "consumer.run({" it worked but only after the entire script ended.

How can I force it to run and to wait for all messages to be fetched?

kafkajs apache-kafka node.js javascript
1 Answers

The only thing that worked for me was to add a pause after the run() finction. I must say I'm not happy about this solution since I'm forcing my side to wait fixed 5 seconds regardless to the results.

I'll be happy for a better solutions - but here it is: I have added a delay function:

    function delay(time) {
    return new Promise(resolve => setTimeout(resolve, time));
}

And then , after the run "await consumer.run(..." and before "consumer.disconnect()" I called it Like this:

    await consumer.run({
        eachMessage: async ({ message }) => {
             messages.push(message)
             console.log('RECEIVED MESSAGE', JSON.parse(message.value), message.offset);
          }
        })
           
        })
/*Here is where the code waits for messages to be pulled  */
       await delay(5000)
       consumer.disconnect();
Gil Allen0
Related