Offset Management

[sumanyu]

I'm new to Kafka and trying to understand offset management.

As far as I understand, a consumer group implicitly uses ZooKeeper to store offsets per partition. Are the starting offsets for partitions automatically set to 0 for each consumer group?

After initializing SimpleConsumer, I tried to grab a message using kafka_consumer.get_message(get_partition_info=True), but it failed with the error message kafka failed with error code=1 (OFFSET OUT_OF_RANGE).

Do I need to explicitly set the starting offsets using kafka_consumer.seek(0, 0)?

[sumanyu]

UPDATE: I'm ultimately trying to make Kafka work with Storm (petrel lib). Any pointers on integrating the two would be awesome.

[wizzat]

Are you using Kafka 0.8.1 or Kafka 0.8.0? While the offsets are ultimately stored in Zookeeper, Kafka 0.8.1 allows direct management of offsets. There are several pull requests which address this recent change in various ways. There's some discussion of how to go about handling the prevalent 0.8.0 and the new 0.8.1.

[sumanyu]

I was originally using 0.8.0 until I realized it didn't support commits. I'm using 0.8.1 now.

[wizzat]

The current version of kafka-python doesn't support resuming from offsets without a patch or two. There's a lot of well merited discussion about how to handle multiple versions of Kafka with semi-incompatible APIs, and there are multiple pull requests open which implement various suggestions on how to solve this.

I believe your current options are:

• Always process from the head of the stream
• Always process from the tail of the stream
• Maintain offset management manually using Kazoo or some other data store of your own devising
• Use a fork of kafka-python which supports what you need.

Full disclosure: I am maintaining a fork of kafka-python which supports multiple versions of kafka, has been through fairly extensive QA at KIXEYE, and is in production. The fork is at: https://github.com/wizzat/kafka-python/tree/easykafka. This fork will likely become obsolete once the above discussion is resolved in some manner.

[asdfsx]

I store offsets in a local file, and I'm using 0.8.0. Now I'm trying to move to 0.8.1

[wizzat]

This should be fixed in #158.

[dpkp]

Pls reopen if this isn't fixed in 0.9.1 / master.