bpot/poseidon — pull request #19 by bpot (2013-09-26 21:24:42+00:00)
Title: Consumer crashes sometimes depending on max_bytes size

I sometimes see the following exception on the consumer side

Poseidon::Protocol::ProtocolStruct::DecodingError: Error while reading message in Poseidon::Protocol::MessageWithOffsetStruct (NoMethodError: undefined method `unpack' for nil:NilClass))
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1 /gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:188:in `rescue in block in read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:182:in `block in read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:181:in `each'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:181:in `read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:155:in `read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/message.rb:36:in `read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:221:in `read_type'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:204:in `read_member'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:183:in `block in read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:181:in `each'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:181:in `read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:155:in `read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/message_set.rb:12:in `read'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:221:in `read_type'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:212:in `read_member'
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/protocol/protocol_struct.rb:183:in `block in read'
... 25 levels...
from /Users/bwessels/.rbenv/versions/1.9.3-p286/lib/ruby/gems/1.9.1/gems/poseidon-0.0.3/lib/poseidon/partition_consumer.rb:90:in `fetch'

This is an intermittent problem with the default max_bytes of (1024*1024)

From debugging it seems when the consumer requests messages and it hits the max_bytes AND it gets a few extra bytes of the next message then the decoding fails as it thinks there is another message but it runs out data for the decode.

I reproduced this by producing 2 short messages then I consume with a max_bytes of just over the size of one of the messages and it fails. If I change the max_bytes from 362 to 373 the messages are correctly handled. This may be similar to
https://github.com/bpot/poseidon/issues/9 where kafka delivers partial messages.

To reproduce do the following
1) Make sure the kafka queue is completely empty (delete the kafka-log files and restart kafka)
2) From command line produce the following 2 messages

$KAFKA/kafka-console-producer.sh --broker-list localhost:9092 --topic cenx.elm.inventory.updates
{:action :create, :object {"cos_end_point_id" 18, "self_id" 17, "id" 18, "base_class" "CosEndPointsCosEndPoint", "self_id_public_id" "CosEndPoint:e9fde2def9673e5d13388d7a268c244c12e30c5c5b534c536cc114b2ff9c9667", "cos_end_point_id_public_id" "CosEndPoint:37b5588002235c06cb36bcb7e2b2337ce8f959adb182067906c2f50461e366ea"}, :changed {}}
{:action :create, :object {"cos_end_point_id" 18, "self_id" 17, "id" 18, "base_class" "CosEndPointsCosEndPoint", "self_id_public_id" "CosEndPoint:e9fde2def9673e5d13388d7a268c244c12e30c5c5b534c536cc114b2ff9c9667", "cos_end_point_id_public_id" "CosEndPoint:37b5588002235c06cb36bcb7e2b2337ce8f959adb182067906c2f50461e366ea"}, :changedYES {}}

3) From a rails console consume the messages

queue = Poseidon::PartitionConsumer.new("updater_consumer", "localhost", 9092, "cenx.elm.inventory.updates", 0, 0)
loop do
messages = queue.fetch(:max_bytes => 362, :offset =>:earliest_offset)
if messages.present?
messages.each do |message|
puts message
end
else
sleep(1)
end
end

4) exception will occur - Using a message size of 373 works