node.js - 主题不存在错误 : The topic(s) xxx do not exist in kafka-node?

Question

我正在使用kafka构建两个消费者客户端，如下所示的代码，但它给了我一个错误，说

TopicsNotExistError: The topic(s) DEVICE_BINDED do not exist
    at new TopicsNotExistError (/Users/xisizhe/Documents/projects/mean-master/node_modules/_kafka-node@3.0.0@kafka-node/lib/errors/TopicsNotExistError.js:11:9)
    at /Users/xisizhe/Documents/projects/mean-master/node_modules/_kafka-node@3.0.0@kafka-node/lib/client.js:464:43

如何解决？

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

function ConnectionProvider() {
    this.getConsumer = function (topic_name) {
          console.log('consumer consumer',topic_name);
          this.client = new kafka.Client("localhost:2181");
          this.kafkaConsumerConnection = new kafka.Consumer(this.client,[ { topic: topic_name, partition: 0 }]);
          this.kafkaConsumerConnection.on('ready', function () {
            console.log('consumer ready!')
          })
      return this.kafkaConsumerConnection;
    };

    //Code will be executed when we start Producer
    this.getProducer = function () {

        if (!this.kafkaProducerConnection) {
            this.client = new kafka.Client("localhost:2181");
            var HighLevelProducer = kafka.HighLevelProducer;
            //this.kafkaProducerConnection = new HighLevelProducer(this.client);
            this.kafkaProducerConnection = new kafka.Producer(this.client);
            var self = this;
            this.kafkaProducerConnection.on('ready', function () {
                logger.debug('producer ready');

                self.kafkaProducerConnection.on('error', function (err) {
                    console.error("Error " + err)
                })
            });

        }
        return this.kafkaProducerConnection;
    };
}


const device_add_consumer  = connection.getConsumer(config.kafka.topic.DEVICE_ADDED);

const device_binded_consumer  = connection.getConsumer(config.kafka.topic.DEVICE_BINDED);

Answer 1

do I have to explicitly create the topic for the client

是的，你应该这样做。否则你会得到代理设置的任何默认值(如果你只使用本地主机，你可能没有调整)

I assume it can create the topic through this code automatically

这不是一个安全的假设。 Kafka 服务器属性定义是否自动创建主题，并且可以禁用该主题。

您需要通过 Node 或使用 Kafka 安装中的 kafka-topics --create CLI 工具手动创建它