default partition offset = 0 breaks when first log segment is rolled

[dpkp]

When not using auto-commit, the consumer object will set the default offset to 0 for all partitions:
https://github.com/mumrah/kafka-python/blob/master/kafka/consumer.py#L119

This is an absolute offset and will only work so long as the first message is still in an active log segment. As soon as it gets rolled by the server, the default consumer will no longer be able to fetch messages without a manual seek.

I think we should probably query the server for the smallest valid offset rather than assuming offset 0 is still there.

[wizzat]

I wonder how you'd go about writing a test for that. A unit test may be as good as we get on that one. But yeah, I totally agree. Arguably, the right answer is actually to start reading from the tail of the topic instead of the head.

[leporid]

I think I just ran into this, in a bad way. I have a log.retention.hours=168, and send_offset_request() reports back [OffsetResponse(topic='events', partition=0, error=0, offsets=(774096,))].

Of course, I'm getting BrokerResponseError on the first fetch of a MultiProcessConsumer restart after an unrelated exception -- and our system needs to be live -- so I'm going to try to debug this.

[dpkp]

leporid -- workaround is to use consumer.seek(0,0) to jump to the first available offset after initializing your consumer. or seek(0,2) to jump to the last available offset.

[leporid]

Thanks for the hint.

I considered that, but, sensibly, seek() is only a method on SimpleConsumer, not MultiProcessConsumer.

My solution was this: At line 118 in consumer.py, in Consumer.__init__():

 for partition in partitions:
     req = OffsetFetchRequest(topic, partition)
     (commit_offset,) = self.client.send_offset_fetch_request(group, [req],
             callback=get_or_init_offset_callback,
             fail_on_error=False
        )
    (earliest_offset,) = self.client.send_offset_request([OffsetRequest(topic, partition, -2, 1)])[0].offsets
    self.offsets[partition] = max(commit_offset, earliest_offset)

Does this make sense?

[leporid]

My private monkey patch disappeared when I upgraded to 0.9.2, and this is still open. May I attempt to fix this and generate a PR? (Solution quite similar to above, but also noticed that auto_commit == False flag also defaults partition offsets to 0, so I can fix that too.)

[dpkp]

Fixed in #296 by reseting partition offsets on offset out of range errors