Kafka-node: No events being emitted from client or producer

[RobertHerhold]

Bug Report

Environment

• Node version: 8.1.0
• Kafka-node version: 1.6.2
• Kafka version: 0.10.2.1

Include Sample Code to reproduce behavior

const kafka = require('kafka-node');

const kafkaClient = new kafka.Client('<my-remote-url>:2181');
const kafkaProducer = new kafka.Producer(kafkaClient);

kafkaClient.on('error', function (err) {
    logger.error('Kafka Client error: ', err);
});

kafkaClient.on('ready', () => console.log('ready'));

kafkaClient.on('connect', function () {
    console.log('connected');

    kafkaClient.loadMetadataForTopics([], function (error, results) {
        if (error) {
            return console.error(error);
        }
        console.log(results);
    });
});

kafkaProducer.on('error', function (err) {
    logger.error('Kafka Producer error: ', err);
});

// And then later on...

kafkaProducer.send([{
    topic: 'test',
    messages: ['test message']
}], function (err) {
    console.log(err);
});

Output

None of my event listeners result in anything being logged, so I'm not sure where to begin. Using env DEBUG="kafka-node:*" when running my server doesn't seem to provide any additional output.

{ BrokerNotAvailableError
    at new BrokerNotAvailableError (/Users/robert/projects/my-project/node_modules/kafka-node/lib/errors/BrokerNotAvailableError.js:11:9)
    at Client.loadMetadataForTopics (/Users/robert/projects/my-project/node_modules/kafka-node/lib/client.js:348:15)
    at Client.send (/Users/robert/projects/my-project/node_modules/kafka-node/lib/client.js:515:10)
    at /Users/robert/projects/my-project/node_modules/kafka-node/lib/client.js:220:10
    at /Users/robert/projects/my-project/node_modules/kafka-node/node_modules/async/lib/async.js:52:16
    at Object.async.forEachOf.async.eachOf (/Users/robert/projects/my-project/node_modules/kafka-node/node_modules/async/lib/async.js:236:30)
    at Object.async.forEach.async.each (/Users/robert/projects/my-project/node_modules/kafka-node/node_modules/async/lib/async.js:209:22)
    at Client.sendProduceRequest (/Users/robert/projects/my-project/node_modules/kafka-node/lib/client.js:218:9)
    at Producer.BaseProducer.send (/Users/robert/projects/my-project/node_modules/kafka-node/lib/baseProducer.js:123:10)
    at Object.<anonymous> (/Users/robert/projects/my-project/ws-handler.js:37:15)
    at Module._compile (module.js:569:30)
    at Object.Module._extensions..js (module.js:580:10)
    at Module.load (module.js:503:32)
    at tryModuleLoad (module.js:466:12)
    at Function.Module._load (module.js:458:3)
    at Module.require (module.js:513:17)
    at require (internal/module.js:11:18)
    at Object.<anonymous> (/Users/robert/projects/my-project/app.js:139:5)
    at Module._compile (module.js:569:30)
    at Object.Module._extensions..js (module.js:580:10)
    at Module.load (module.js:503:32)
    at tryModuleLoad (module.js:466:12)
    at Function.Module._load (module.js:458:3)
    at Function.Module.runMain (module.js:605:10)
    at startup (bootstrap_node.js:158:16)
    at bootstrap_node.js:575:3 message: 'Broker not available' }

And when I use kafkacat to confirm that everything is set up right:

kafkacat -b <my-remote-url>:9092 -L

Metadata for all topics (from broker 0: <my-remote-url>:9092/0):
 1 brokers:
  broker 0 at <my-remote-url>:9092
 1 topics:
  topic "test" with 1 partitions:
    partition 0, leader 0, replicas: 0, isrs: 0

It seems that everything is set up right, with 1 broker that the kafka-node client should be able to connect to. Are there any issues that you can see from what I've posted?

[RobertHerhold]

I should also mention that I gave it plenty of time before executing the kafkaProducer.send function; it wasn't being called immediately like my code sample shows

[krystianity]

You are most likely not even connected to your broker.. which is why you see the BrokerNotAvailable Error when calling .send()..
As of now, kafka-node requires a zookeeper to connect to the broker - since your ready and connect events are not firing, the client is not able to connect to zk in the first place.

The zk module wont fire/log any events when its unable to connect it will just try to reconnect silently.
Using kafkacat you directly connect to the broker and omit the procedure a kafka-node has to run through.

[RobertHerhold]

@krystianity I also tried the branch discussed in #666 to bypass ZK, but I have the same issue. Any ideas?

[krystianity]

Can't you use zk for now? Until #666 moves into master/production?

[RobertHerhold]

I can, I just have to figure out why it can't connect to zk then :/

[RobertHerhold]

@krystianity I got it figured out, it was just a zookeeper config change. Thanks for the help

[toml7]

@RobertHerhold I am having the same issues . Can you please tell me what was the config change that was made so that I can try it out?

[jaiswarvipin]

The title and content is not matching.
Title say: On event is not getting emited by kafak-node lib...
and Body Say: Broker is not available.

Lets solve thing one by one...
1) Broker is not available.
Complete details mention at #160.

2) Producer and Consumer not emitting the ready event
Please try below code

var intCounter = 104;

    setInterval(() => {
        setMessageInTopic('topic'+intCounter, function(response){
            echo("Producer Response"+response);
            if(response){
                echo("Consumer init");
                getMessageFromKafak(response,function(pStrMessage){
                    echo ("------------------------Consumer message --------------------");
                    echo(pStrMessage);
                });
            }else{
                echo("Error");
            }
        });
        intCounter++; 
    },5000);



function setMessageInTopic(pStrTopicName, callback){
    const kafka = require('kafka-node');

    try {
      const Producer = kafka.Producer;
      const client = new kafka.KafkaClient({kafkaHost:"localhost:9094"});
      const producer = new Producer(client);
      const kafka_topic =  pStrTopicName;
      console.log(kafka_topic);
      let payloads = [
        {
          topic: kafka_topic,
          messages: {'name':'vipin'}
        }
      ];

      producer.on('ready', async function() {
        let push_status = producer.send(payloads, (err, data) => {
          if (err) {
            console.log('[kafka-producer -> '+kafka_topic+']: broker update failed');
            callback(false);
          } else {
            console.log('[kafka-producer -> '+kafka_topic+']: broker update success');
            callback(pStrTopicName);
          }
        });
      });

      producer.on('error', function(err) {
        console.log(err);
        console.log('[kafka-producer -> '+kafka_topic+']: connection errored');
        callback(false);
        ///throw err;
      });
    }
    catch(e) {
      console.log(e);
      callback(false);
    }
}

function getMessageFromKafak(pStrTopicName, callback){
    try{
        var kafka = require('kafka-node');
        //var HighLevelProducer = kafka.HighLevelProducer;
        var Consumer = kafka.Consumer;
        var client = new kafka.KafkaClient({kafkaHost:"localhost:9094"});

        let consumer = new Consumer(
            client,
            [{ topic: pStrTopicName}],
            {
              autoCommit: true,
              fetchMaxWaitMs: 1000,
              fetchMaxBytes: 1024 * 1024,
              encoding: 'utf8',
              fromOffset: false
            }
          );

        echo("------------------RESPONE MESSAGE PROCESS "+pStrTopicName+" ----------------------")
        consumer.on('message', function (message) {
            echo("------------------RESPONE MESSAGE  ----------------------")
            console.log(message);
            consumer.close();
            client.close();
            return callback(message);
        }).on('error', function (message) {
            echo("------------------RESPONE MESSAGE ERROR ----------------------")
            console.log(message);
            consumer.close();
            client.close();
            return callback(message);
        });
    }catch(e) {
      console.log(e);
    }
}

Hope that helps.

Thanks & Regards
Jaiswar Vipin Kumar R.