mumrah/kafka-python — pull request #148 by wizzat (2014-03-25 18:20:02+00:00)
Title: Make seek(); commit(); work without commit discarding the seek change

Once upon a time I was moving between Kafka 0.8.0 and Kafka 0.8.1 rather freely. As such, I forked kafka-python. In particular, I've enabled offset management via the Kafka API, added a bunch of performance tweaks, and in general made it work. The results are visible here: https://github.com/wizzat/kafka-python/tree/easykafka.

However, during a load test we ran Kafka out of disk space. In response to this, we set the TTL on the cluster to about an hour, which deleted all the data out of the cluster. The offsets the application had stored in Zookeeper (via the Kafka server) were now wrong - and worse, didn't exist. This caused the application to raise lots of OFFSET_OUT_OF_RANGE errors.

In response to these OFFSET_OUT_OF_RANGE errors, I devised a brilliant work around. At the start of the application, I would advance the topic offsets something like this:

def seek_tail(self):
logging.info('Seeking to TAIL of topic %s', self.topic)
logging.info('Previous offsets:')
self.log_offsets()

self.consumer.seek(0, 2)
self.consumer.commit()

logging.info('New offsets')
self.log_offsets()

However, as it turns out, however, no data actually went through the application - and thus the commit was lost due to the following line:

# short circuit if nothing happened. This check is kept outside
# to prevent un-necessarily acquiring a lock for checking the state
if self.count_since_commit == 0:
return

I thought about extracting commit() into two methods: commit() which delegated to force_commit(), however it seemed a smaller change to make seek() count as a count_since_commit. Thus, I propose a patch.

Let me know if you'd prefer to go with commit() and force_commit(), or perhaps commit(partitions=None, force=False).