Make seek(); commit(); work without commit discarding the seek change

[wizzat]

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).

[dpkp]

only side-effect i can think of is that it could change the auto_commit counting. but I think that's fine. i say merge. will let others comment.

[rdiomar]

count_since_commit could be considered the number of times the consumer's offset changed since the last commit rather than the number of messages consumed, so I'm okay with this, but I feel like if auto_commit was True, some might expect seeking to do an immediate commit. Does anyone have a strong opinion about this?

[wizzat]

That makes a lot of sense and I'm tentatively +1 to it. I will implement this sometime today unless someone objects.

[wizzat]

Ok. I've updated the PR to commit when auto_commit.