Safely handle first consume with auto.offset.reset latest and reaching end of partition

[herndlm]

Moved from arnaud-lb/php-rdkafka#529 here where it most likely belongs and/or gets more attention I hope.

Hi,

first of all, I'm a Kafka beginner, so I assume the 2 questions I have are some really basic things where I'm not fully understanding some concepts yet. Maybe somebody can help me there or point me to some useful resources in that regard.

Basically they are potentially quite similar since they deal with timeouts I'm getting. And the consumer code is very close to https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka.examples-high-level-consumer.html

The idea is to have a consuming CLI application that is triggered e.g. every 5m and consumes messages.

1. We'd like to consume a topic with auto.offset.reset being set to latest since we don't care about previous messages. But if I do that it seems the consumer just "waits" with the given timeout and, in our case, leads to a timeout error then if no message comes in while consuming. At least as long as there is no offset commited yet which is the case when we consume the first time. Is this expected behaviour? How would I correctly consume for the first time then?
2. If we reach the end of the partitions it looks like the next consume call is waiting for the next message to come in and if there isn't any coming, given the specified timeout via argument, it leads to a timeout error again. shouldn't this be an end-of-partition error instead? Currently I'm having problems differentiating between end of partition / no new messages atm and a "real" timeout because of e.g. network reasons.

Thank you for bearing with me with those very very basic questions.. :)

[herndlm]

Sorry, I assume without logs nobody can help much. So, I just ran the consumer again and consumed everything from 6 partitions. I set the timeout for the next consume to some very high level and in the logs I can see that it does something in a loop, it seems to be fetching data and doing something with offsets forever. Extract from the log

%7|1657617103.519|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://b0-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b0-<KAFKA-DOMAIN>:9092/0: Fetch topic <TOPIC> [1] at offset 32127 (v4)
%7|1657617103.519|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://b0-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b0-<KAFKA-DOMAIN>:9092/0: Fetch 1/1/1 toppar(s)
%7|1657617103.519|SEND|rdkafka#consumer-1| [thrd:sasl_ssl://b0-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b0-<KAFKA-DOMAIN>:9092/0: Sent FetchRequest (v11, 138 bytes @ 0, CorrId 4)
%7|1657617103.519|RECV|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b4-<KAFKA-DOMAIN>:9092/4: Received SaslAuthenticateResponse (v0, 8 bytes, CorrId 3, rtt 34.02ms)
%7|1657617103.519|SASL|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b4-<KAFKA-DOMAIN>:9092/4: Received SASL frame from broker (0 bytes)
%7|1657617103.519|STATE|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b4-<KAFKA-DOMAIN>:9092/4: Broker changed state AUTH_REQ -> UP
%7|1657617103.519|BROADCAST|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: Broadcasting state change
%7|1657617103.519|METADATA|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: Skipping metadata refresh of 1 topic(s): connected: already being requested
%7|1657617103.519|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b4-<KAFKA-DOMAIN>:9092/4: Fetch topic <TOPIC> [5] at offset 32336 (v4)
%7|1657617103.519|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b4-<KAFKA-DOMAIN>:9092/4: Fetch 1/1/1 toppar(s)
%7|1657617103.520|SEND|rdkafka#consumer-1| [thrd:sasl_ssl://b4-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b4-<KAFKA-DOMAIN>:9092/4: Sent FetchRequest (v11, 138 bytes @ 0, CorrId 4)
%7|1657617103.833|RECV|rdkafka#consumer-1| [thrd:sasl_ssl://b5-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b5-<KAFKA-DOMAIN>:9092/5: Received FetchResponse (v11, 110 bytes, CorrId 8, rtt 533.94ms)
%7|1657617103.833|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://b5-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b5-<KAFKA-DOMAIN>:9092/5: Topic <TOPIC> [0] MessageSet size 0, error "Success", MaxOffset 33038, LSO 33038, Ver 4/4
%7|1657617103.833|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://b5-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b5-<KAFKA-DOMAIN>:9092/5: Fetch topic <TOPIC> [0] at offset 33038 (v4)
%7|1657617103.833|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://b5-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b5-<KAFKA-DOMAIN>:9092/5: Fetch 1/1/1 toppar(s)
%7|1657617103.834|SEND|rdkafka#consumer-1| [thrd:sasl_ssl://b5-<KAFKA-DOMAIN>:9092/boot]: sasl_ssl://b5-<KAFKA-DOMAIN>:9092/5: Sent FetchRequest (v11, 138 bytes @ 0, CorrId 9)
%7|1657617103.858|CGRPOP|rdkafka#consumer-1| [thrd:main]: Group "<CONSUMER-GROUP>" received op OFFSET_COMMIT in state up (join-state steady)
%7|1657617103.858|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic <TOPIC> [0]: stored offset -1001, committed offset 33038: not including in commit
%7|1657617103.858|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic <TOPIC> [1]: stored offset -1001, committed offset 32127: not including in commit
%7|1657617103.858|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic <TOPIC> [2]: stored offset -1001, committed offset 32151: not including in commit
%7|1657617103.858|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic <TOPIC> [3]: stored offset -1001, committed offset 32882: not including in commit
%7|1657617103.858|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic <TOPIC> [4]: stored offset -1001, committed offset 32400: not including in commit
%7|1657617103.858|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic <TOPIC> [5]: stored offset -1001, committed offset 32336: not including in commit
%7|1657617103.858|COMMIT|rdkafka#consumer-1| [thrd:main]: OffsetCommit for 6 partition(s) in join-state steady: manual: returned: Local: No offset stored
%7|1657617103.858|DUMP|rdkafka#consumer-1| [thrd:main]: Assignment dump (started_cnt=6, wait_stop_cnt=0)
%7|1657617103.858|DUMP_ALL|rdkafka#consumer-1| [thrd:main]: List with 6 partition(s):
%7|1657617103.858|DUMP_ALL|rdkafka#consumer-1| [thrd:main]:  <TOPIC> [0] offset STORED
%7|1657617103.858|DUMP_ALL|rdkafka#consumer-1| [thrd:main]:  <TOPIC> [1] offset STORED
%7|1657617103.858|DUMP_ALL|rdkafka#consumer-1| [thrd:main]:  <TOPIC> [2] offset STORED
%7|1657617103.858|DUMP_ALL|rdkafka#consumer-1| [thrd:main]:  <TOPIC> [3] offset STORED
%7|1657617103.858|DUMP_ALL|rdkafka#consumer-1| [thrd:main]:  <TOPIC> [4] offset STORED
%7|1657617103.858|DUMP_ALL|rdkafka#consumer-1| [thrd:main]:  <TOPIC> [5] offset STORED
%7|1657617103.858|DUMP_PND|rdkafka#consumer-1| [thrd:main]: List with 0 partition(s):
%7|1657617103.858|DUMP_QRY|rdkafka#consumer-1| [thrd:main]: List with 0 partition(s):
%7|1657617103.858|DUMP_REM|rdkafka#consumer-1| [thrd:main]: List with 0 partition(s):
%7|1657617103.858|ASSIGNDONE|rdkafka#consumer-1| [thrd:main]: Group "<CONSUMER-GROUP>": assignment operations done in join-state steady (rebalance rejoin=false)

Messages like [0] MessageSet size 0, error "Success", MaxOffset 33038, LSO 33038, Ver 4/4 were scrolling by from time to time regularly. With the exact same MaxOffset value.

And if the timeout is reached, I get an error response

string(28) "RD_KAFKA_RESP_ERR__TIMED_OUT"
int(-185)

but instead I would have been expecting a RD_KAFKA_RESP_ERR__PARTITION_EOF.

[nick-zh]

So regarding your first question:

1. Not caring about old messages but then still expecting some is kind of confusing to me. auto.offset.reset will be used and applied the first time you connect or when you have no commited offset. So if you want to skip old messages, but get all future ones, you might want to connect with LATEST but commit the current offset (i think this is what you are trying to achieve)
2. If you want to get EOF you need to enable it, the default behaviour was changed (see example)

I hope that helps to clear things up.

[herndlm]

oh what a coincidence, I was looking into this since half an hour or so again. I really appreciate your help, thx!

1. yeah, that sounds exactly what I want to do! I just need to figure out how to do that with the high-level consumer. Will look into it.
2. oh god, how could I miss that one.. yep, that's also what I was looking for. Seems to be working. thx again!

[nick-zh]

Regarding 1.
You probably commit after you've read a message, so you can just trigger / wait for one message
There are cli tools that can help
Or in the php-rdkafka doc you can use one of the functions to get the latest offset for reach partition and commit it

Hope this answers you question ✌️

[herndlm]

Jfyi wanted to tell you that with setting enable.partition.eof the initial consuming behaviour with auto.offset.reset set to latest changed too, which made sense. And I was able to set the offsets to latest by getting the max values via https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka-kafkaconsumer.querywatermakoffsets.html 🎉

What most likely threw me off in the beginning were the php-rdkafka examples. All of them handle EOF but none set enable.partition.eof. I think I'm gonna open a PR to adapt that. Thx again!