Support for the librdkafka stats_cb

[trink]

https://github.com/edenhill/librdkafka/wiki/Statistics
mozilla-services/hindsight#137 (comment)

[trink]

filename                = "kafka.lua"
output_limit            = 8 * 1024 * 1024
brokerlist              = "localhost:9092" -- see https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L2205

-- In balanced consumer group mode a consumer can only subscribe on topics, not topics:partitions.
-- The partition syntax is only used for manual assignments (without balanced consumer groups).
topics                  = {"test_output"}

-- https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#global-configuration-properties
consumer_conf = {
    log_level = 7,
    ["statistics.interval.ms"] = 1000, 
    ["group.id"] = "test_group2", -- must always be provided (a single consumer is considered a group of one
    -- in that case make this a unique identifier)
    ["message.max.bytes"] = output_limit,
}

-- https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#topic-configuration-properties
topic_conf = {
    -- ["auto.commit.enable"] = true, -- cannot be overridden
    -- ["offset.store.method"] = "broker, -- cannot be overridden
}

-- Specify a module that will decode the raw data and inject the resulting message.
--decoder_module = "decoders.heka.protobuf" -- default

1513716844161305891 [info] rdkafka#consumer-1 { "name": "rdkafka#consumer-1", "type": "consumer", "ts":22941906481, "time":1513716844, "replyq":0, "msg_cnt":0, "msg_size":0, "msg_max":0, "msg_size_max":0, "simple_cnt":0, "metadata_cache_cnt":1, "brokers":{ "localhost:9092/bootstrap": { "name":"localhost:9092/bootstrap", "nodeid":-1, "state":"UP", "stateage":3001005, "outbuf_cnt":0, "outbuf_msg_cnt":0, "waitresp_cnt":0, "waitresp_msg_cnt":0, "tx":4, "txbytes":118, "txerrs":0, "txretries":0, "req_timeouts":0, "rx":4, "rxbytes":218, "rxerrs":0, "rxcorriderrs":0, "rxpartial":0, "zbuf_grow":0, "buf_grow":0, "wakeups":11, "int_latency": { "min":0, "max":0, "avg":0, "sum":0, "cnt":0 }, "rtt": { "min":0, "max":0, "avg":0, "sum":0, "cnt":0 }, "throttle": { "min":0, "max":0, "avg":0, "sum":0, "cnt":0 }, "toppars":{ } } , "ubuntu:9092/0": { "name":"ubuntu:9092/0", "nodeid":0, "state":"UP", "stateage":2998179, "outbuf_cnt":0, "outbuf_msg_cnt":0, "waitresp_cnt":1, "waitresp_msg_cnt":0, "tx":67, "txbytes":4709, "txerrs":0, "txretries":0, "req_timeouts":0, "rx":66, "rxbytes":8307431, "rxerrs":0, "rxcorriderrs":0, "rxpartial":0, "zbuf_grow":0, "buf_grow":0, "wakeups":154, "int_latency": { "min":0, "max":0, "avg":0, "sum":0, "cnt":0 }, "rtt": { "min":297, "max":105617, "avg":13067, "sum":561892, "cnt":43 }, "throttle": { "min":0, "max":0, "avg":0, "sum":0, "cnt":42 }, "toppars":{ "test_output-0": { "topic":"test_output", "partition":0} } } }, "topics":{ "test_output": { "topic":"test_output", "metadata_age":2983, "partitions":{ "0": { "partition":0, "leader":0, "desired":true, "unknown":false, "msgq_cnt":0, "msgq_bytes":0, "xmit_msgq_cnt":0, "xmit_msgq_bytes":0, "fetchq_cnt":0, "fetchq_size":0, "fetch_state":"active", "query_offset":0, "next_offset":4601219, "app_offset":4601219, "stored_offset":4601219, "commited_offset":4579309, "committed_offset":4579309, "eof_offset":4601219, "lo_offset":-1001, "hi_offset":4601219, "consumer_lag":0, "txmsgs":0, "txbytes":0, "msgs": 21910, "rx_ver_drops": 0 } , "-1": { "partition":-1, "leader":-1, "desired":false, "unknown":false, "msgq_cnt":0, "msgq_bytes":0, "xmit_msgq_cnt":0, "xmit_msgq_bytes":0, "fetchq_cnt":0, "fetchq_size":0, "fetch_state":"none", "query_offset":0, "next_offset":0, "app_offset":-1001, "stored_offset":-1001, "commited_offset":-1001, "committed_offset":-1001, "eof_offset":-1001, "lo_offset":-1001, "hi_offset":-1001, "consumer_lag":-1, "txmsgs":0, "txbytes":0, "msgs": 0, "rx_ver_drops": 0 } } } } , "cgrp": { "rebalance_age": 2980, "rebalance_cnt": 2, "assignment_size": 1 }}

[trink]

#204