Friday 9 June 2017

How to add timestamp to kafka 10 messages with nodejs

I re-wrote a Kafka producer node in node-red using nodejs. I noted Kafka10 added feature about attaching timestamp on messages sent to producers but I found just java examples. How can I add timestamp to messages using nodejs ?

In the following I report how producer is implemented in nodejs:

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});
}

On google I found just this:link

but actually I don't understand how to integrate in my code (what's Avro?)



via Andrea Giordano

No comments:

Post a Comment