Handle normal offset exceptions:

This happens when one of the following is true.

1. You have multiple queues and one host crashes. The new host is
brougt back to life but the offset in zookeeper is wrong. So it will
throw an offset exception.

2. Similarly when you unmount the drive where kafka is writing to.

This patch tries to ask kafka what the latest offset is in this case.
If none is return, it uses zero and assumes the queue is empty. This
has to be the case, otherwise kafka would return a different offset
that is not zero.

This will at most replay a whole offset block. I think they are at
512M. That's just how kafka handles offsets. It will give you the
latest offset by chunk where chunks are of 512M in size.