Restore pull request #376 from zendesk/dasch/message-timestamps

This reverts commit d03eae0.

Author: dasch

Diff for: circle.yml

@@ -9,7 +9,7 @@ machine:
 dependencies:
   pre:
     - docker -v
-    - docker pull ches/kafka:0.9.0.1
+    - docker pull ches/kafka:0.10.0.0
     - docker pull jplock/zookeeper:3.4.6
     - gem install bundler -v 1.9.5
 

Diff for: lib/kafka/fetch_operation.rb

@@ -90,6 +90,7 @@ def execute
                 topic: fetched_topic.name,
                 partition: fetched_partition.partition,
                 offset: message.offset,
+                create_time: message.create_time,
               )
             }
 

Diff for: lib/kafka/fetched_message.rb

@@ -16,12 +16,16 @@ class FetchedMessage
     # @return [Integer] the offset of the message in the partition.
     attr_reader :offset
 
-    def initialize(value:, key:, topic:, partition:, offset:)
+    # @return [Time] the timestamp of the message.
+    attr_reader :create_time
+
+    def initialize(value:, key:, topic:, partition:, offset:, create_time:)
       @value = value
       @key = key
       @topic = topic
       @partition = partition
       @offset = offset
+      @create_time = create_time
     end
   end
 end

Diff for: lib/kafka/protocol/fetch_request.rb

@@ -30,6 +30,10 @@ def api_key
         FETCH_API
       end
 
+      def api_version
+        2
+      end
+
       def response_class
         Protocol::FetchResponse
       end

Diff for: lib/kafka/protocol/fetch_response.rb

@@ -38,11 +38,14 @@ def initialize(name:, partitions:)
 
       attr_reader :topics
 
-      def initialize(topics: [])
+      def initialize(topics: [], throttle_time_ms: 0)
         @topics = topics
+        @throttle_time_ms = throttle_time_ms
       end
 
       def self.decode(decoder)
+        throttle_time_ms = decoder.int32
+
         topics = decoder.array do
           topic_name = decoder.string
 
@@ -68,7 +71,7 @@ def self.decode(decoder)
           )
         end
 
-        new(topics: topics)
+        new(topics: topics, throttle_time_ms: throttle_time_ms)
       end
     end
   end

Diff for: lib/kafka/protocol/message.rb

@@ -6,15 +6,16 @@ module Protocol
 
     # ## API Specification
     #
-    #     Message => Crc MagicByte Attributes Key Value
+    #     Message => Crc MagicByte Attributes Timestamp Key Value
     #         Crc => int32
     #         MagicByte => int8
     #         Attributes => int8
+    #         Timestamp => int64, in ms
     #         Key => bytes
     #         Value => bytes
     #
     class Message
-      MAGIC_BYTE = 0
+      MAGIC_BYTE = 1
 
       attr_reader :key, :value, :codec_id, :offset
 
@@ -71,14 +72,15 @@ def self.decode(decoder)
         end
 
         attributes = message_decoder.int8
+        timestamp = message_decoder.int64
         key = message_decoder.bytes
         value = message_decoder.bytes
 
         # The codec id is encoded in the three least significant bits of the
         # attributes.
         codec_id = attributes & 0b111
 
-        new(key: key, value: value, codec_id: codec_id, offset: offset)
+        new(key: key, value: value, codec_id: codec_id, offset: offset, create_time: Time.at(timestamp/1000.0))
       end
 
       private
@@ -102,6 +104,7 @@ def encode_without_crc
 
         encoder.write_int8(MAGIC_BYTE)
         encoder.write_int8(@codec_id)
+        encoder.write_int64((@create_time.to_f*1000).to_i)
         encoder.write_bytes(@key)
         encoder.write_bytes(@value)
 

Diff for: lib/kafka/protocol/produce_request.rb

@@ -40,6 +40,10 @@ def api_key
         PRODUCE_API
       end
 
+      def api_version
+        2
+      end
+
       def response_class
         requires_acks? ? Protocol::ProduceResponse : nil
       end

Diff for: lib/kafka/protocol/produce_response.rb

@@ -11,19 +11,21 @@ def initialize(topic:, partitions:)
       end
 
       class PartitionInfo
-        attr_reader :partition, :error_code, :offset
+        attr_reader :partition, :error_code, :offset, :timestamp
 
-        def initialize(partition:, error_code:, offset:)
+        def initialize(partition:, error_code:, offset:, timestamp:)
           @partition = partition
           @error_code = error_code
           @offset = offset
+          @timestamp = timestamp
         end
       end
 
-      attr_reader :topics
+      attr_reader :topics, :throttle_time_ms
 
-      def initialize(topics: [])
+      def initialize(topics: [], throttle_time_ms: 0)
         @topics = topics
+        @throttle_time_ms = throttle_time_ms
       end
 
       def each_partition
@@ -43,13 +45,16 @@ def self.decode(decoder)
               partition: decoder.int32,
               error_code: decoder.int16,
               offset: decoder.int64,
+              timestamp: Time.at(decoder.int64/1000.0),
             )
           end
 
           TopicInfo.new(topic: topic, partitions: partitions)
         end
 
-        new(topics: topics)
+        throttle_time_ms = decoder.int32
+
+        new(topics: topics, throttle_time_ms: throttle_time_ms)
       end
     end
   end

Diff for: spec/consumer_spec.rb

@@ -58,6 +58,7 @@
           topic: "greetings",
           partition: 0,
           offset: 13,
+          create_time: Time.now,
         )
       ]
     }
@@ -182,6 +183,7 @@
           topic: "greetings",
           partition: 0,
           offset: 13,
+          create_time: Time.now,
         )
       ]
     }

Diff for: spec/fake_broker.rb

@@ -35,6 +35,7 @@ def produce(messages_for_topics:, required_acks:, timeout:)
             partition: partition,
             error_code: error_code_for_partition(topic, partition),
             offset: message_set.messages.size,
+            timestamp: (Time.now.to_f*1000).to_i,
           )
         }
       )

Diff for: spec/test_cluster.rb

@@ -18,7 +18,7 @@ class TestCluster
   }
 
   DOCKER_HOSTNAME = URI(DOCKER_HOST).host
-  KAFKA_IMAGE = "ches/kafka:0.9.0.1"
+  KAFKA_IMAGE = "ches/kafka:0.10.0.0"
   ZOOKEEPER_IMAGE = "jplock/zookeeper:3.4.6"
   KAFKA_CLUSTER_SIZE = 3