How to capture record-lag-max metric for consumer?

[kasturichavan]

We want to scale our kafka consumer with HPA in kubernetes based on kafka custom metric record-lag. Is there a method in confluent-kafka-python that exposes the metrics ? How can we get this data?
Below is the fetch metrics list:
https://docs.confluent.io/current/kafka/monitoring.html#fetch-metrics

Checklist

Please provide the following information:

• confluent-kafka-python version (confluent_kafka.version(0.11.5)):

[edenhill]

See consumer_lag here: https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md#partitions

confluent-kafka-python/examples/consumer.py

Line 71 in 590937f

conf['stats_cb'] = stats_cb

[kasturichavan]

I set the "statistics.interval.ms": 10 for testing. So now consumer.poll() should trigger the stats_cb. But this isnt returning any data ?? Will stats data be returned in poll function ??

[edenhill]

You need to define a stats_cb callback too, and call poll() at regular intervals to trigger the stats.
(a stats interval of 10ms might be a tad low).

[kasturichavan]

where do we pass the stats_cb ? Sorry my question could be dumb. But based on code specified above, stats_cb seems to be already defined. Can this be passed as a field to consumer init() call ? Docs dnt have that specified as a field consumer init accepts.

[edenhill]

confluent-kafka-python/examples/consumer.py

Line 29 in 590937f

def stats_cb(stats_json_str):

[kasturichavan]

@edenhill After implementing above, i do get back stats.
Noticed following issues:

1. Sometimes topics field & toppars field are empty & sometimes they hold values from which i can extract the consumer_lag.https://docs.confluent.io/current/clients/librdkafka/md_STATISTICS.html

2. The stats that get gathered are consumer specific right? So it only includes partitions that are assigned to this consumer. Or does it include info of all partitions in the topic that are assigned to all consumers and its lags ???

[edenhill]

Stats will only be collected for the currently assigned/consumed partitions.

[kasturichavan]

@edenhill But why do the stats hold empty values for topics field for a single consumer in consumergroup assigned to a topic with 2 partitions. These stats are collected via a consumer poll() and hence it will always be pointing to a currently assigned partition.

[edenhill]

@kasturichavan Can you paste your stats json object?