Don't assume there's a timestamp on messages

Old messages may not have one.

Author: dasch

lib/kafka/protocol/message.rb

@@ -88,7 +88,10 @@ def self.decode(decoder)
         # attributes.
         codec_id = attributes & 0b111
 
-        new(key: key, value: value, codec_id: codec_id, offset: offset, create_time: Time.at(timestamp / 1000.0))
+        # The timestamp will be nil if the message was written in the Kafka 0.9 log format.
+        create_time = timestamp && Time.at(timestamp / 1000.0)
+
+        new(key: key, value: value, codec_id: codec_id, offset: offset, create_time: create_time)
       end
 
       private

spec/fixtures/message-0.9-format

@@ -13,4 +13,17 @@
 
     expect(Kafka::Protocol::Message.decode(decoder)).to eq message
   end
+
+  it "decodes messages written in the 0.9 format" do
+    data = File.open("spec/fixtures/message-0.9-format")
+
+    decoder = Kafka::Protocol::Decoder.new(data)
+    message = Kafka::Protocol::Message.decode(decoder)
+
+    expect(message.key).to eq "xx"
+    expect(message.value).to eq "yolo"
+
+    # Messages didn't have timestamps back in the 0.9 days.
+    expect(message.create_time).to eq nil
+  end
 end