apache-kafka - BrokerNotAvailableError : Could not find the leader Exception while Spark Streaming

Question

我在 NodeJS 中编写了 Kafka Producer，在 Java Maven 中编写了 Kafka Consumer。我的主题是“test”，它是由以下命令创建的:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

NodeJS中的生产者:

var kafka = require('kafka-node');
var Producer = kafka.Producer;
var Client = kafka.Client;
var client = new Client('localhost:2181');
var producer = new Producer(client);

producer.on('ready', function () {
    producer.send([
        { topic: 'test', partition: 0, messages: ["This is the zero message I am sending from Kafka to Spark"], attributes: 0},
        { topic: 'test', partition: 1, messages: ["This is the first message I am sending from Kafka to Spark"], attributes: 0},
        { topic: 'test', partition: 2, messages: ["This is the second message I am sending from Kafka to Spark"], attributes: 0}
        ], function (err, result) {
        console.log(err || result);
        process.exit();
    });
});

当我从 NodeJS 生产者发送两条消息时，它成功被 Java 消费者消费。但是当我从 NodeJS 生产者发送三个或更多消息时，它给了我以下错误:

{ [BrokerNotAvailableError: 找不到领导者] 消息:'找不到领导者' }

我想问一下，如何将 LEADER 设置为主题“测试”中的任何消息。或者该问题的解决方案应该是什么。

Answer 1

而不是 partition使用 partitions (复数键名)。

例如:

producer.on('ready', function () {
  producer.send([
    { topic: 'test', partitions: 0, messages: ["This is the zero message I am sending from Kafka to Spark"], attributes: 0},
    { topic: 'test', partitions: 1, messages: ["This is the first message I am sending from Kafka to Spark"], attributes: 0},
    { topic: 'test', partitions: 2, messages: ["This is the second message I am sending from Kafka to Spark"], attributes: 0}
  ], function (err, result) {
    console.log(err || result);
    process.exit();
  });
});