Ability to seek in topic

[rjbell4]

I'd like to be able to seek to an offset within the topic. At the moment, it's useful when testing a Kafka consumer, as it makes it easier to iterate over the same messages with different versions of the consumer.

I believe this implies calling rd_kafka_seek, which doesn't seem to be exposed by Kafka::Librd.

[trinitum]

librdkafka exposes 2 different consumer APIs: simple and high-level. High-level one provides support for consumer groups and is the one that is supported by Kafka::Librd. rd_kafka_seek is part of the simple API and in order to make use of it I would have to add support for all the simple API which is not something I'm planning to do (but I'll be happy to accept pull request that adds such support). In high-level API, there's rd_kafka_commit function that allows you to commit offsets to the broker, perhaps it can satisfy your needs?

[fecundf]

Jumping in a few years later with the same issue. I'm trying to follow the suggestion to use rd_kafka_commit via Kafka::Librd but have a couple of blockers crafting a call to $consumer->commit that winds to the start or skips to the end.

1. What offset to use? The topic may not begin at 0. I can't find any calls exposed in Kafka::Librd to get the start/end offsets for a topic & partition
2. I try calling commit with 0, or with a large number, or even with an offset that I know is inside the range of valid offsets, and that value is reflected in "committed", but it doesn't change what I retrieve with the next "consumer_poll". How does one get the consumer to respect the new offset?

debug session below

  DB<7> x $consumer->committed([{topic=>'my_topic',partition => 0}],1000)
0  ARRAY(0x7fcb50e17cb0)
   0  HASH(0x7fcb50534820)
      'offset' => 1385
      'partition' => 0
      'topic' => 'my_topic'
  DB<8> x $consumer->commit([{topic=>'my_topic',partition => 0, offset=>2**31}])
0  0
  DB<9> x $consumer->committed([{topic=>'my_topic',partition => 0}],1000)
0  ARRAY(0x7fcb50534d78)
   0  HASH(0x7fcb505347f0)
      'offset' => 2147483648
      'partition' => 0
      'topic' => 'my_topic'
  DB<10> x $msg=$consumer->consumer_poll(1000);
0  Kafka::Librd::Message=SCALAR(0x7fcb50e17b30)
   -> 140511268232240
  DB<11> p join ' ', map eval('$msg->'.$_), qw[err topic partition offset key payload timestamp]
0 my_topic 0 1385 key 1 {"t1":Wed Jan 20 16:32:19 2021} 1611178339054

the above shows that the consumer begins at offset 1385, then "commit" succeeds in setting the offset to a large int, but the next call to consumer_poll still retrieves from 1385. It doesn't matter what offset I commit, consumer_poll keeps advancing one message at a time regardless.

[fecundf]

A possible solution-
create new RD_KAFKA_CONSUMER with config {'enable.auto.commit'=> 'false', 'auto.offset.reset'=>'earliest'}

Do not subscribe, instead call assign with a tplist having desired offset for each topic/partition, use offset=0 for beginning. eg

$kafka->assign([{offset=>0,topic=>'re-read-me',partition=>0}])

then $kafka->consumer_poll reads the earlier entries.

Maintainers, can you document that offset is valid in the @tplist for assign? Would save some trial & error. If there are any other options those hashes control, document them too!

side question/comment, for boolean config options like enable.auto.commit, use the names 'true/false' or numbers 0/1? Would be good to add that in the documentation. I'm not entirely sure what the underlying librdkafka lib expects and any translation the Perl module does.