Add ability to specify the record timestamp

[consulthys]

Since Kafka 0.10.0.0, it is now possible to include a timestamp in Kafka records.

It would be great if it was possible to also specify this in the kafka output plugin with the message_timestamp setting (e.g.). The value of that setting could either be:

• the name of a field (e.g. "@timestamp") whose value can be resolved to a long value
• a sprintf-style field (e.g. "%{second_timestamp}000") whose interpolated value can be resolved to a long value

A sample configuration would look like this:

input {
  stdin {}
}
output {
  kafka {
    bootstrap_servers => "localhost:9092"
    topic_id => "test"
    message_key => "%{key_field}"
    message_timestamp => "@timestamp"
    # or like this
    # message_timestamp => "%{@second_timestamp}000"
  }
}

The record would then need to be created using the full-fledge ProducerRecord constructor:

msg_ts = get_timestamp_value_and_coerce_to_long(@message_timestamp) unless @message_timestamp.nil?
record = org.apache.kafka.clients.producer.ProducerRecord.new(event.sprintf(@topic_id), null, msg_ts, event.sprintf(@message_key), data)

[consulthys]

@guyboertje what about this one?

[guyboertje]

@consulthys I would need to get a second opinion. /cc @suyograo

[neerav-salaria-guavus]

We would also like to have this one, as of now all logs have the same TimeStamp in kafka which is CreateTime irrespective of each record having its own @timestamp field.

[John-Athan]

The missing functionality as described here is still missing from the plugin, right? Is there a chance for it to still be solved inside this repository? Else, I would open a new issue here https://github.com/logstash-plugins/logstash-integration-kafka

[0x1346614]

@John-Athan Have you opened the issue?

[John-Athan]

@0x1346614 no, sorry, the topic got less relevant for us.