consumer manual commit not working

[Ghilteras]

Hi, when I run my consumer script I disable auto commits because in my use case there are some messages that need to be processed and if we got no errors then we can remove them from the queue hence the manual commit bit.

But when I manually commit the offset it is still there on next iteration of my script; of course if I re-enable auto-commit it's gone..

this is my code:

from kafka import KafkaConsumer
from kafka.structs import OffsetAndMetadata

consumer = KafkaConsumer(bootstrap_servers='192.168.33.10:9092', consumer_timeout_ms=1000, enable_auto_commit=False)
consumer.subscribe(['my-topic', 'another-topic'])

for message in consumer:
    print (message)
    meta = consumer.partitions_for_topic(message.topic)
    partition = consumer.assignment().pop()
    offsets = OffsetAndMetadata(message.offset, meta)
    options = {partition: offsets}
    consumer.commit(offsets=options)
consumer.close()

any hints?

[dpkp]

assignment().pop() is wrong. That could give you any partition, but you want the particular messag e partition. Use TopicPartition(message.topic, message.partition)

[dpkp]

Also you can leave meta as b'' unless you have some use for that field yourself

[Ghilteras]

Thanks, that commits everything but the latest though, it's not that the consumer gets killed or something, it exits cleanly so I would expect the latest message to be committed as well

[dpkp]

Kafka commits are typically the offset of the next message to fetch, not the last message consumed. So try offset + 1 when you commit an offset

[Ghilteras]

yea with offset +1 I seem I can commit manually just fine, thanks

[dpkp]

We could probably make the documentation better for this, but I'm going to close because the OP issue has been resolved.

[tkaymak]

Is this still the right way to do manual commits?