breaking: Nest decorated fields under @metadata (#213)

* breaking: Nest decorated fields under `@metadata`

Fixes #198

* add docs about the metdata information

Author: Suyog Rao

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 7.0.0
+  - Breaking: Nest the decorated fields under `@metadata` field to avoid mapping conflicts with beats.
+    Fixes #198, #180
+
 ## 6.3.4
   - Fix an issue that led to random failures in decoding messages when using more than one input thread
 

docs/index.asciidoc

@@ -64,6 +64,19 @@ For more information see http://kafka.apache.org/documentation.html#theconsumer
 
 Kafka consumer configuration: http://kafka.apache.org/documentation.html#consumerconfigs
 
+==== Metadata fields
+
+The following metadata from Kafka broker are added under the `[@metadata]` field:
+
+* `[@metadata][kafka][topic]`: Original Kafka topic from where the message was consumed.
+* `[@metadata][kafka][consumer_group]`: Consumer group
+* `[@metadata][kafka][partition]`: Partition info for this message.
+* `[@metadata][kafka][offset]`: Original record offset for this message.
+* `[@metadata][kafka][key]`: Record key, if any.
+* `[@metadata][kafka][timestamp]`: Timestamp when this message was received by the Kafka broker.
+
+Please note that `@metadata` fields are not part of any of your events at output time. If you need these information to be 
+inserted into your original event, you'll have to use the `mutate` filter to manually copy the required fields into your `event`.
 
 [id="plugins-{type}s-{plugin}-options"]
 ==== Kafka Input Configuration Options

lib/logstash/inputs/kafka.rb

@@ -254,12 +254,12 @@ def thread_runner(logstash_queue, consumer)
             codec_instance.decode(record.value.to_s) do |event|
               decorate(event)
               if @decorate_events
-                event.set("[kafka][topic]", record.topic)
-                event.set("[kafka][consumer_group]", @group_id)
-                event.set("[kafka][partition]", record.partition)
-                event.set("[kafka][offset]", record.offset)
-                event.set("[kafka][key]", record.key)
-                event.set("[kafka][timestamp]", record.timestamp)
+                event.set("[@metadata][kafka][topic]", record.topic)
+                event.set("[@metadata][kafka][consumer_group]", @group_id)
+                event.set("[@metadata][kafka][partition]", record.partition)
+                event.set("[@metadata][kafka][offset]", record.offset)
+                event.set("[@metadata][kafka][key]", record.key)
+                event.set("[@metadata][kafka][timestamp]", record.timestamp)
               end
               logstash_queue << event
             end

logstash-input-kafka.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-input-kafka'
-  s.version         = '6.3.4'
+  s.version         = '7.0.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = 'This input will read events from a Kafka topic. It uses the high level consumer API provided by Kafka to read messages from 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/integration/inputs/kafka_spec.rb

@@ -134,9 +134,9 @@ def thread_it(kafka_input, queue)
         wait(timeout_seconds).for {queue.length}.to eq(num_events)
         expect(queue.length).to eq(num_events)
         event = queue.shift
-        expect(event.get("kafka")["topic"]).to eq("logstash_topic_plain")
-        expect(event.get("kafka")["consumer_group"]).to eq(group_id_3)
-        expect(event.get("kafka")["timestamp"]).to be >= start
+        expect(event.get("[@metadata][kafka][topic]")).to eq("logstash_topic_plain")
+        expect(event.get("[@metadata][kafka][consumer_group]")).to eq(group_id_3)
+        expect(event.get("[@metadata][kafka][timestamp]")).to be >= start
       ensure
         t.kill
         t.join(30_000)