autocommit offset does not work if i use high level api

[ago3x]

I use the high level api(rd_kafka_subscribe and rd_kafka_consumer_poll) to consume messages,but when i restart the programe it will consume the message from offset 0.

[edenhill]

Are you calling consumer_close() before exiting?

Can you reproduce this with examples/rdkafka_consumer_example ?

[ago3x]

i don't call the consumer_close(),the offset record by consumer_close()?

[edenhill]

You need to call consume_close() to properly shut down your consumer and let it perform final offset commits, group leaves, etc.

[ago3x]

Have api to record offset by myself?

[edenhill]

You may commit offsets yourself with rd_kafka_commit().

[ago3x]

can't auto commit?

[edenhill]

auto commit is enabled by default, but you must call consumer_close() before exiting.

[ago3x]

I think this is not good,if my program crash the offset will not commit.

[edenhill]

If auto commit is enabled (enable.auto.commit=true) offsets will be committed periodically (auto.commit.interval.ms) to the broker.

When exiting an application must also call consumer_close() to commit any offsets waiting to be committed from the last interval.

This is a pay off between performance and consistency.
If you commit the offset for each message you will get full consistency and no message replays on restart, but it will affect performance negatively.

[DavidLiuXh]

I set 'enable.auto.commit=true' and 'auto.commit.interval.ms', but can't periodically committed.

[edenhill]

What do you set ..interval.ms to?
How many messages are you receiving?
How long is your program running?
Are you calling consumer_close()?

[ago3x]

mee too

[ago3x]

auto.commit.interval.ms=4000
about 20 messages
2min
no

[edenhill]

Run your program with debug=topic and see if it says anything about offset commits.

Also, what is the reason for not calling consumer_close()?
librdkafka relies on it to perform a proper clean shut down.

[ago3x]

RDKAFKA-7-TOPIC: rdkafka#consumer-0: New local topic: test1
RDKAFKA-7-DESP: rdkafka#consumer-0: Adding desired topic test1 [0]
RDKAFKA-7-OFFSET: rdkafka#consumer-0: PC-ZHENGJIYUE.hikvision.com:9092/0: Offset
FetchRequest(v1) for 1 partition(s)
RDKAFKA-7-OFFSET: rdkafka#consumer-0: Topic test1 [0]: setting default offset -1
001
RDKAFKA-7-CONSUMER: rdkafka#consumer-0: Start consuming test1 [0] at offset TAIL
(76843841185970498)
RDKAFKA-7-OP: rdkafka#consumer-0: test1 [0] received op FETCH_START (v2) in fetc
h-state none
RDKAFKA-7-FETCH: rdkafka#consumer-0: Start fetch for test1 [0] in state none at
offset TAIL(76843841185970498) (v2)
RDKAFKA-7-PARTSTATE: rdkafka#consumer-0: Partition test1 [0] changed fetch state
none -> offset-query
RDKAFKA-7-OFFSET: rdkafka#consumer-0: test1 [0]: offset reset (at offset TAIL(76
843841185970498)) to TAIL(76843841185970498): update: Success
RDKAFKA-7-OFFSET: rdkafka#consumer-0: test1 [0]: no current leader for partition
, starting offset query timer for offset TAIL(76843841185970498)
RDKAFKA-7-STATE: rdkafka#consumer-0: Topic test1 changed state unknown -> exists

RDKAFKA-7-PARTCNT: rdkafka#consumer-0: Topic test1 partition count changed from
0 to 1
RDKAFKA-7-BRKDELGT: rdkafka#consumer-0: Broker PC-ZHENGJIYUE.hikvision.com:9092/
0 is now leader for topic test1 [0] with 0 messages (0 bytes) queued
RDKAFKA-7-METADATA: rdkafka#consumer-0: PC-ZHENGJIYUE.hikvision.com:9093/1: Requ
ested topic test1 seen in metadata
RDKAFKA-7-OFFSET: rdkafka#consumer-0: Topic test1 [0]: timed offset query for TA
IL(76843841185970498)
RDKAFKA-7-OFFREQ: rdkafka#consumer-0: PC-ZHENGJIYUE.hikvision.com:9092/0: Partit
ion test1 [0]: querying for logical offset TAIL(76843841185970498) (opv 2)
RDKAFKA-7-OFFSET: rdkafka#consumer-0: PC-ZHENGJIYUE.hikvision.com:9092/0: Offset
Request (-1) for topic test1 [0]
RDKAFKA-7-PARTSTATE: rdkafka#consumer-0: Partition test1 [0] changed fetch state
offset-query -> offset-wait
RDKAFKA-7-OFFSET: rdkafka#consumer-0: Offset -76843841185972498 request for test
1 [0] returned offset 25 (25)
RDKAFKA-7-OFFSET: rdkafka#consumer-0: OffsetReply for topic test1 [0]: offset 25
: adjusting for OFFSET_TAIL(76843841185970498): effective offset 0
RDKAFKA-7-PARTSTATE: rdkafka#consumer-0: Partition test1 [0] changed fetch state
offset-wait -> active
RDKAFKA-7-FETCHADD: rdkafka#consumer-0: PC-ZHENGJIYUE.hikvision.com:9092/0: Adde
d test1 [0] to fetch list (1 entries)