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

[herndlm]

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]

fyi I also posted this to confluentinc/librdkafka#3905 where I added logs

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.

[herndlm]

this was answered via confluentinc/librdkafka#3905 (comment)

I missed setting enable.partition.eof to true and the initial offset had to be comited manually to latest.