Added ability to specify timestamp in message

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 6.1.5
+  - Add ability to specify message timestamp (#124)
+
 ## 6.1.4
   - Fix a bug where consumer was not correctly setup when `SASL_SSL` option was specified.
 

lib/logstash/outputs/kafka.rb

@@ -98,6 +98,10 @@ class LogStash::Outputs::Kafka < LogStash::Outputs::Base
   config :max_request_size, :validate => :number, :default => 1048576
   # The key for the message
   config :message_key, :validate => :string
+  # The timestamp for the message. This can either be:
+  #  - a field containing a timestamp value
+  #  - an sprintf-style field containing a timestamp value
+  config :message_timestamp, :validate => :string
   # the timeout setting for initial metadata request to fetch topic metadata.
   config :metadata_fetch_timeout_ms, :validate => :number, :default => 60000
   # the max time in milliseconds before a metadata refresh is forced.
@@ -178,11 +182,20 @@ def register
     @producer = create_producer
     @codec.on_event do |event, data|
       begin
-        if @message_key.nil?
-          record = org.apache.kafka.clients.producer.ProducerRecord.new(event.sprintf(@topic_id), data)
-        else
-          record = org.apache.kafka.clients.producer.ProducerRecord.new(event.sprintf(@topic_id), event.sprintf(@message_key), data)
+        partition = nil
+        key = event.sprintf(@message_key) unless @message_key.nil?
+
+        timestamp = nil
+        unless @message_timestamp.nil?
+          if event.include? @message_timestamp
+            ts = event.get(@message_timestamp)
+            timestamp = ts.is_a?(LogStash::Timestamp) ? ts.time.to_i : ts.to_i
+          else
+            timestamp = event.sprintf(@message_timestamp).to_i
+          end
         end
+
+        record = org.apache.kafka.clients.producer.ProducerRecord.new(event.sprintf(@topic_id), partition, timestamp, key, data)
         @producer.send(record)
       rescue LogStash::ShutdownSignal
         @logger.debug('Kafka producer got shutdown signal')

logstash-output-kafka.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-output-kafka'
-  s.version         = '6.1.4'
+  s.version         = '6.1.5'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = 'Output events to a Kafka topic. This uses the Kafka Producer API to write messages to a topic on the broker'
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/unit/outputs/kafka_spec.rb

@@ -6,7 +6,7 @@
 describe "outputs/kafka" do
   let (:simple_kafka_config) {{'topic_id' => 'test'}}
   let (:event) { LogStash::Event.new({'message' => 'hello', 'topic_name' => 'my_topic', 'host' => '172.0.0.1',
-                                      '@timestamp' => LogStash::Timestamp.now}) }
+                                      '@timestamp' => LogStash::Timestamp.now, 'other_timestamp' => 1487669184685}) }
 
   context 'when initializing' do
     it "should register" do
@@ -34,7 +34,7 @@
     it 'should support Event#sprintf placeholders in topic_id' do
       topic_field = 'topic_name'
       expect(org.apache.kafka.clients.producer.ProducerRecord).to receive(:new)
-        .with("my_topic", event.to_s)
+        .with("my_topic", nil, nil, nil, event.to_s)
       expect_any_instance_of(org.apache.kafka.clients.producer.KafkaProducer).to receive(:send)
       kafka = LogStash::Outputs::Kafka.new({'topic_id' => "%{#{topic_field}}"})
       kafka.register
@@ -43,13 +43,31 @@
 
     it 'should support field referenced message_keys' do
       expect(org.apache.kafka.clients.producer.ProducerRecord).to receive(:new)
-        .with("test", "172.0.0.1", event.to_s)
+        .with("test", nil, nil, "172.0.0.1", event.to_s)
       expect_any_instance_of(org.apache.kafka.clients.producer.KafkaProducer).to receive(:send)
       kafka = LogStash::Outputs::Kafka.new(simple_kafka_config.merge({"message_key" => "%{host}"}))
       kafka.register
       kafka.receive(event)
     end
-
+
+    it 'should support timestamp from specified message field' do
+      expect(org.apache.kafka.clients.producer.ProducerRecord).to receive(:new)
+        .with("test", nil, event.get('@timestamp').time.to_i, nil, event.to_s)
+      expect_any_instance_of(org.apache.kafka.clients.producer.KafkaProducer).to receive(:send)
+      kafka = LogStash::Outputs::Kafka.new(simple_kafka_config.merge({"message_timestamp" => "@timestamp"}))
+      kafka.register
+      kafka.receive(event)
+    end
+
+    it 'should support field referenced message_timestamp' do
+      expect(org.apache.kafka.clients.producer.ProducerRecord).to receive(:new)
+        .with("test", nil, event.get('other_timestamp').to_i, nil, event.to_s)
+      expect_any_instance_of(org.apache.kafka.clients.producer.KafkaProducer).to receive(:send)
+      kafka = LogStash::Outputs::Kafka.new(simple_kafka_config.merge({"message_timestamp" => "%{other_timestamp}"}))
+      kafka.register
+      kafka.receive(event)
+    end
+
     it 'should raise config error when truststore location is not set and ssl is enabled' do
       kafka = LogStash::Outputs::Kafka.new(simple_kafka_config.merge({"ssl" => "true"}))
       expect { kafka.register }.to raise_error(LogStash::ConfigurationError, /ssl_truststore_location must be set when SSL is enabled/)