Kakfa Transaction Time in Newrelic NodeJS?

How can I measure kafka transaction time in newrelic? Currently newrelic is not showing my kafka transaction call. Is there a way that I can measure kafka transaction time like redis or mysql.

With my best knowledge, it can be done with custom tracers using createTracer, but how I can use newrelic in the kafka file.

Here is my file structure:

var nr = require('newrelic');
var express  = require('express');
app = express();
var port = process.env.PORT || 8000;
var bodyParser = require('body-parser');
var cookieParser = require('cookie-parser');
var gs = require('./utils').producer;
app.use(bodyParser.json());     // support json encoded bodies
app.use(bodyParser.urlencoded({ extended: false }));    // support encoded bodies
app.use(cookieParser());

app.listen(port);
console.log('Server started! At http://localhost:' + port);

var statsCollector = function (req,res) {

    var cookies = req.cookies;
    var headres = req.headers;
    var message = req.body;

    var data  = {headers: headers, cookies: cookies, message: message}

    gs.send("vtap_main", data)

    res.status(200).json({"message": "ok", "status": true}).end();

}

app.post('/*', statsCollector);

Here is my kafka file in utils: ./utils/index.js ./utils/kafka.js

var StringProducer = require('kafka-java-bridge').StringProducer;
var BinaryProducer = require('kafka-java-bridge').BinaryProducer;
var HLConsumer = require("kafka-java-bridge").HLConsumer;

var zkUrl = process.env.ZK_URL || 'localhost:2181';

var stringProducer = new StringProducer( { zookeeperUrl: zkUrl } );
var binaryProducer = new BinaryProducer( { zookeeperUrl: zkUrl } );

exports.send = nr.createTracer('kafka', function (topic, messages) {

    if (messages.constructor != Array) messages = [messages];

    messages.forEach(function (msg) {
        if (msg.constructor == String) stringProducer.send(topic, msg);
        else if (msg.constructor == Buffer) binaryProducer.send(topic, msg);
        else if (msg.constructor == Object) stringProducer.send(topic, JSON.stringify(msg));
        else throw "Invalid Message Type!"
    });

});

exports.consumer = function (topics) {

    if (topics.constructor != Array) topics = [topics];
    var consumerOptions = {
            zookeeperUrl: zkUrl,
            groupId: "vtap-processor",
            getMetadata: true,
            topics: topics,
            serverPort: 3042,// Optional
            threadCount: 4,// Optional
            properties: {"rebalance.max.retries": "3"}// Optional
        };

    return new HLConsumer(consumerOptions);

}

exports.Consumer = 

process.on('SIGINT', function() {
    stringProducer.close(function(err){
        binaryProducer.close(function(err) {
            process.exit();
        });
    });
});

How can I use nr from server.js in utils/kafka.js to measure or create the customTrace in newrelic? Is there any other betterway to do this?

via 7H3 IN5ID3R