Kafka input - reset offset to beginning?

[Milan0x4d]

Hello everyone,

I wrote a simple Kafka input test config script:

[SERVICE]
Flush 1
Log_Level info

[INPUT]
Name kafka
Brokers 192.168.1.3:9092, 192.168.1.5:9092
Topics my-test-topic
Group_id my_CG_1
Client_id my_client_app
rdkafka.security.protocol SSL
rdkafka.ssl.certificate.location /home/m/test/cert/my_cert.pem
rdkafka.ssl.key.location /home/m/test/cert/my_key.pem
rdkafka.ssl.ca.location /home/m/test/cert/intermediate.pem

[OUTPUT]
Name stdout
Match *

Fluentbit consumes just fine, when I produce to "my-test-topic". However, I also have a need to re-read messages periodically. I thought "well, this will be simple - just change the "Group_id" to something else; you get a new CG, and bam! you're back to offset 0.

This turned out not to be the case, though. It looks like offsets are remembered by fluentbit, regardless of the CG name? Even with the new CG, I am sent to the end of the topic. I can see newly produced messages, but none are visible from the beginning.

Does anyone know how we could force offsets to go to 0? With standard "kafka-consumer", you have an option "--from-beginning" that does this for us. What would the fluentbit equivalent be?

Side not: I know the default CG is "fluent-bit", so I wonder if this is perhaps the issue? Maybe this is hard-coded and our "group_id" doesn't override it?

[Milan0x4d]

Update - some good news and one outstanding question:

Good news - I blew away the whole setup and started from scratch. Changing CG name in the kafka-input section now reads the messages from the beginning of the offset, as expected. What happened? Mea culpa - I think my old messages expired on the topic over the weekend, which is why I couldn't read anything this morning (nothing was there). Once I recreated everything and ran a battery of tests, I can confirm that changing the CG name in the Fluentbit config file will read the messages from the beginning of the offset. Every time you want to read the topic from the beginning, you must change to a new CG name.

One outstanding question: I'd still like the ability to rewind the offset to 'beginning', with the existing CG.

I tried a setting from librdkafka called "auto offset reset", like so:

[INPUT]
Name kafka
Brokers 192.168.1.3:9092, 192.168.1.5:9092
Topics my-test-topic
Group_id my_CG_1
Client_id my_client_app
rdkafka.security.protocol SSL
rdkafka.ssl.certificate.location /home/m/test/cert/my_cert.pem
rdkafka.ssl.key.location /home/m/test/cert/my_key.pem
rdkafka.ssl.ca.location /home/m/test/cert/intermediate.pem
rdkafka.auto.offset.reset beginning

(The last line was added).

It didn't work as expected: I read the topic with the same CG, but offset wasn't reset to the beginning. Only new messages would show up. I also tried the value of 'earliest', but that didn't work either.

So my revised question: how can one rewind the existing CG to read the topic from the earliest offset?

[agup006]

@Milan0x4d it looks like from rdkafka that the setting enabled still takes into account if an offset store is set. Potentially you can set the store to off and it wiill work as expected?

Action to take when there is no initial offset in offset store or the desired offset is out of range: 'smallest','earliest' - automatically reset the offset to the smallest offset, 'largest','latest' - automatically reset the offset to the largest offset, 'error' - trigger an error (ERR__AUTO_OFFSET_RESET) which is retrieved by consuming messages and checking 'message->err'.
Type: enum value

[Milan0x4d]

Thank you @agup006 , this was perfect!

My settings:

[INPUT]
Name kafka
Brokers 192.168.1.3:9092, 192.168.1.5:9092
Topics my-test-topic
Group_id my_CG_2
Client_id my_client_app
rdkafka.security.protocol SSL
rdkafka.ssl.certificate.location /home/m/test/cert/my_cert.pem
rdkafka.ssl.key.location /home/m/test/cert/my_key.pem
rdkafka.ssl.ca.location /home/m/test/cert/intermediate.pem
rdkafka.enable.auto.offset.store false
rdkafka.auto.offset.reset beginning

With the "enable.auto.offset.store" set to false, I can re-read with the same group and it always consumes topics from the beginning. This was exactly what I wanted. Thank you very much!