Cannot consume messages from beginning when put consume.on('message,callback) inside another callback

[leanhvu1989]

Questions?

I push messages from kafka consumer to mongodb.

• If I put the MongoClient.connect(url, function(err, client) {}) inside callback of consumer.on('message', callback), the consumer is able to get messages from beginning.
• However, if I put the consumer.on('message', callback) inside MongoClient.connect(url, function(err, client) {}), then the consumer not get messages from beginning. I'd like to do this way because I don't want to open connect to mongoDB for each message.
• Check the Troubleshooting / FAQ ?
• Search the issues ?

Bug Report

Environment

• Node version: 10.13.0
• Kafka-node version: 4.1.3
• Kafka version: 2.12

For specific cases also provide

• Number of Brokers: 1
• Number partitions for topic: 1

Include Sample Code to reproduce behavior

// include code here
const kafka = require('kafka-node');
const mongo = require('mongodb');
const assert = require('assert');

const { Consumer, Offset, KafkaClient } = kafka;
const { MongoClient } = mongo;

const topic = 'testprocessing';

const url = 'mongodb://localhost:27017';
const dbName = 'test_kafka_processing';
let mongodb;

const client = new KafkaClient({kafkaHost: 'localhost:9092'});

const topics = [{
    topic: 'testprocessing',
    offset: 0,
    partition: 0
}];

const options = {
    autoCommit: false,
    fetchMaxWaitMs: 1000,
    fetchMaxBytes: 1024 * 1024,
    fromOffset: true
};

MongoClient.connect(url, function(err, client) {
    assert.equal(null, err);
    mongodb = client.db(dbName);

    consumer.on('message', (message) => {
        const collection = mongodb.collection('transaction');
        // Insert some documents
        collection.insertOne(message.value,
            function(err, result) {
                assert.equal(err, null);
                console.log("Inserted message into the collection");
                callback(result);
        });
    });

    consumer.on('error', (err) => {
        console.log('error', err);
    });
});

const consumer = new Consumer(client, topics, options);

Include output with Debug turned on

Thanks for your contribution!

[ghost]

Hi, @leanhvu1989 couple observations:

• you are mixing two syntaxes - () => {} and function(){} they have different behaviors so be careful about that, in fact I believe the issue will be in using function(){} because it does not bind proper context for you, not 100% sure though
• please include debug output, it may help us see the issue more clearly