Possible for consumer to seek to a given offset in the topic?

[varunpatro]

Something like this in the Java API:

https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#seek(org.apache.kafka.common.TopicPartition,%20long)

and in librdkafka:

https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.c#L1735

[mfontanini]

That function is listed in the "legacy consumer" part of librdkafka's code. Do you know if it works with the high level consumer in librdkafka?

Otherwise, you can always use Consumer::assign passing the offsets for every partition.

[varunpatro]

Sorry, I'm not sure if that works in the high level consumer.

I did try using Consumer::assign but it fails.

Here's what I did:

auto tp = TopicPartition("test", 0);
tp.set_offset(TopicPartition::OFFSET_END);

cout << tp.get_partition() << tp.get_offset() << endl; // this gives me 0 and -1

consumer.assign({tp});

auto tpl = consumer.get_assignment();
auto offset = consumer.get_offsets(tpl[0]);
count << get<0>(offset) << std<1>(offset) << endl; // this gives me -1001 and -1001

Questions:

1. I need to get the latest position of a message in a given topic. How do I do this?
2. I need to seek to a given position in a topic. How do I do this?

[mfontanini]

Can't you just subscribe to the topic and use auto.offset.reset to latest? Using assign with a specific offset should work, there's nothing that cppkafka is doing besides forwarding that call to librdkafka.

Also, what do you mean "it fails"?

[mfontanini]

I've used this on some application I wrote and it works as expected. Did you manage to make it work? Just assign and you'll start getting messages for that offset.

[mfontanini]

Closing as this should work as expected. If you couldn't make it work, please re-open.

[EvilBeaver]

Same issue.

cppkafka::TopicPartitionList assignments;
offset = TopicPartition::OFFSET_END;
assignments.push_back({topicName, partition, offset});
consumer->assign(assignments);

cout << consumer->get_offsets(consumer->c->get_assignment()[0]);

always gives an INVALID_OFFSET. And consumer doesn't consume anything.

[mfontanini]

If this were to be an issue, it would be on rdkafka's side. cppkafka is just a very thin layer on top of it.

Have you checked the documentation for that call in rdkafka's side? Quoting it:

If there is no cached offset (either low or high, or both) then RD_KAFKA_OFFSET_INVALID will be returned for the respective offset.

It is possible you're querying the offset too quickly and it's still unknown to the library.

And consumer doesn't consume anything.

Are you producing messages on that topic/partition? Your consumer will be pointing to the end of the log so you need to produce something after you subscribe for the consumer to receive something.

[EvilBeaver]

Of course I'm producing on this topic/partition. And bundled kafka console-consumer can read messages. My consumer can't

[mfontanini]

I've just tried this locally using rdkafka v0.11.4 and it works as expected. I do get INVALID_OFFSET after querying right after calling assign but then I get messages and if I query for the offsets again after I get a message, the query does return the (high) offset correctly.

Are you using the latest version of rdkafka? Again, cppkafka is a super thin layer on top of rdkafka so odds are if there's actually an issue, it's on rdkafka's side and you should create a ticket on their repo.

[EvilBeaver]

Okay, thank you very much! I'll continue digging things

[shurazs]

As far as I could understand, there is a recommendation to use function assign instead of seek.
But assign is quite heavy because are new|delete in its realization.
Is it possible to implement seek function?

[filimonov]

Do you know if it works with the high level consumer in librdkafka?

confluentinc/librdkafka#1925