Php-rdkafka: getting some stats

Created on 10 Feb 2018 · 7Comments · Source: arnaud-lb/php-rdkafka

Hi,
Is it possible to get latest offset by topic without consuming it? For example to calculate the lag of consumer.

Source

remizyaka

👍1

Most helpful comment

@remizyaka Any reason why you choose the local file storage over the broker store method? I.e. the JAVA client from >= 0.9.x uses the broker method exclusively.

Additionally you can use the (undocumented) RdKafka\Conf::setStatsCb(callable $callback) method to get the consumer lag, see https://github.com/edenhill/librdkafka/wiki/Consumer-lag-monitoring for more information.

Here an untested example. You can log the consumer lag regularly (callbacks are invoked on poll repeated after statistics.interval.ms) using a simple closure:

$logger = new NullLogger();
$conf = new RdKafka\Conf();

// Additionally you should add the consumer `group.id` in order to split the
// aggregated logs in your monitoring environment
$conf->setStatsCb(function (RdKafka\Kafka $kafka, string $jsonObject) use ($logger) {
    $decoded = \json_decode($jsonObject, true);
    foreach ($decoded['topics'] as $topic => $topicData) {
        foreach ($topicData['partitions'] as $partitionNo => $partitionInfo) {
            $logger->info($partitionInfo['consumer_lag'], [
                'topic' => $topic,
                'partition' => $partitionNo,
            ]);
        }
    }
});

$consumer = new RdKafka\Consumer($conf);
// start consuming...

EDIT: Another option is setting up this monitoring on the broker, which would be perfectly doable when using the broker offset storing method.

tPl0ch on 12 Mar 2018

👍3

All 7 comments

@remizyaka This is possible with librdkafka, and there is already an open issue about this, see #78

I am currently working on adding this to php-rdkafka.

@arnaud-lb This issue could be closed in favour of #78

tPl0ch on 12 Mar 2018

👍1

@tPl0ch actually issue that you mentioned is about opposite approach. Getting STORED offset(latest message read by consymer). I can already get latest stored offset when I'm using local files storage method.
What I really need is to get latest offset on server(real number of messages). To calculate lag of consumer(OFFSET_ON_SERVER - STORED_OFFSET). How can I get it without consuming the topic?

remizyaka on 12 Mar 2018

@remizyaka Any reason why you choose the local file storage over the broker store method? I.e. the JAVA client from >= 0.9.x uses the broker method exclusively.

Here an untested example. You can log the consumer lag regularly (callbacks are invoked on poll repeated after statistics.interval.ms) using a simple closure:

$logger = new NullLogger();
$conf = new RdKafka\Conf();

// Additionally you should add the consumer `group.id` in order to split the
// aggregated logs in your monitoring environment
$conf->setStatsCb(function (RdKafka\Kafka $kafka, string $jsonObject) use ($logger) {
    $decoded = \json_decode($jsonObject, true);
    foreach ($decoded['topics'] as $topic => $topicData) {
        foreach ($topicData['partitions'] as $partitionNo => $partitionInfo) {
            $logger->info($partitionInfo['consumer_lag'], [
                'topic' => $topic,
                'partition' => $partitionNo,
            ]);
        }
    }
});

$consumer = new RdKafka\Consumer($conf);
// start consuming...

EDIT: Another option is setting up this monitoring on the broker, which would be perfectly doable when using the broker offset storing method.

tPl0ch on 12 Mar 2018

👍3

@tPl0ch need more control over stored offset to process some fail situations, so commiting it manually, and sometimes reseting it manually to needed point. When for example server resets the queue and stored offset bigger than available on server. Or when consumer for some reason fails while processing some message. Or when I need to replay messages from given offset.
Thanks for info on consumer lag monitoring, will implement it this way

remizyaka on 12 Mar 2018

@remizyaka

There are hi_offset and lo_offset in the statistics callback (example above), which could tell you "I'm mostly caught up" vs "I'm far behind", but not the exact position on the brokers at the time of query.

There is rd_kafka_query_watermark_offsets in librdkafka, but that requires a network round-trip to the broker every time it is called, so it definitely has quite a cost and would need to be balanced against batch sizing and other factors in high trough-put scenarios. But this would be the only way to get the information from the broker.

Additionally, this method is currently not implemented in php-rdkafka, but I can see why clients of the module would want to use it.

@arnaud-lb what is your opinion on exposing rd_kafka_query_watermark_offsets? Do you think it makes sense to add it?

tPl0ch on 12 Mar 2018

hello, i tried to use RD_KAFKA_OFFSET_STORED,but when i start ,it throws this error:'RdKafka\Exception' with message 'Local: Invalid argument or configuration',code like this:
`

   $rk = new \RdKafka\Consumer();
    $rk->setLogLevel(LOG_DEBUG);       
    $rk->addBrokers(Yii::$app->params['kafka']['ip'] );
    $topicConf = new \RdKafka\TopicConf();
    $topicConf->set('auto.commit.enable', 0);//
    $topic = $rk->newTopic("example2", $topicConf);
    $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

`
I dont't know what the wrong is, i'm new to kafka,hope that i can get u replied, thanks!!