bpot/poseidon — pull request #69 by bkao (2014-10-09 20:33:48+00:00)
Title: added a set_offset() method to partition_consumer.rb

When Kafka has a corrupt message poseidon gets stuck attempting to read the same offset ad infinitum. We needed to be able to skip ahead (presumably past a corrupt message or disk corruption) so we could implement logic like this:

loop do
docs = @kafka.fetch
break if docs.size > 0
break if @kafka.next_offset >= @kafka.highwater_mark

# We can get stuck due to a corrupt message in Kafka. Kafka's
# response (possibly poseidon) doesn't distinguish between no data
# and bad data. Either way we simply get back an empty fetch. To
# get around this, we attempt to advance the offset and re-fetch.
# If offset is past the latest offset, Kafka will throw an error
# and we'll revert the offset.
@kafka.set_offset(@kafka.next_offset + 1)
end