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