如何使用nodejs为kafka 10消息添加时间戳

Question

我使用 nodejs 在 node-red 中重新编写了一个 Kafka 生产者节点。 我注意到 Kafka10 添加了有关在发送给生产者的消息上附加时间戳的功能，但我发现只是 java 示例。 如何使用 nodejs 为消息添加时间戳？

下面我报告nodejs中producer是如何实现的：

function kafkaAdvancedNode(config) {
    RED.nodes.createNode(this,config);
    var topic = config.topic;
    var partition = config.partition;
    var clusterZookeeper = config.zkquorum;
    var debug = (config.debug == "debug");
    var node = this;
    var kafka = require('kafka-node');
    var HighLevelProducer = kafka.HighLevelProducer;
    var Client = kafka.Client;
    var topics = config.topics;
    var client = new Client(clusterZookeeper);

    try {
        this.on("input", function(msg) {
            var payloads = [];

            // check if multiple topics
            if (topics.indexOf(",") > -1){
                var topicArry = topics.split(',');

                for (i = 0; i < topicArry.length; i++) {
                    payloads.push({topic: topicArry[i], messages: msg.payload});
                }
            }
            else {
                if(partition == "" || !partition)
                    payloads = [{topic: topics, messages: msg.payload}];
                else
                    payloads = [{topic: topics, messages: msg.payload, partition: partition}];
            }

            producer.send(payloads, function(err, data){
                if (err){
                    node.error(err);
                }
                node.log("Message Sent: " + data);
            });
        });
    }
    catch(e) {
        node.error(e);
    }
    var producer = new HighLevelProducer(client);
    this.status({fill:"green",shape:"dot",text:"connected to "+clusterZookeeper});
}

在谷歌上我发现了这个：link

但实际上我不明白如何在我的代码中集成（什么是 Avro？）

Answer 1

我认为 kafka-node 不支持 0.10 Kafka 协议，因此它不太可能支持消息时间戳。

出于这个（和其他）原因，我使用 node-rdkafka 编写了类似的内容并将其发布为 node-red-contrib-rdkafka。见https://github.com/hjespers/node-red-contrib-rdkafka

更新：我刚刚发布了 node-red-contrib-rdkafka 的新版本 (0.2.0)，它支持 0.10“事件时间”或“处理时间”时间戳以及动态主题、键和分区。