bpot/poseidon — pull request #68 by bkao (2014-10-09 18:43:47+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:

docs = @kafka.fetch

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

if docs.empty?
curr_offset = @kafka.next_offset
begin
@kafka.set_offset(curr_offset + 1)
docs = @kafka.fetch
rescue Poseidon::Errors::OffsetOutOfRange	=> e
@kafka.set_offset(curr_offset)
sleep 5
end
end