diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index a0333c8ef44b..7185c89a6c0b 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -74,6 +74,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta1...master[Check the HEAD di
 *Filebeat*
 
 - Add PostgreSQL module with slowlog support. {pull}4763[4763]
+- Add Kafka log module. {pull}4885[4885]
 
 *Heartbeat*
 
diff --git a/filebeat/docs/fields.asciidoc b/filebeat/docs/fields.asciidoc
index 4e1cf9673fca..0b677ee0073b 100644
--- a/filebeat/docs/fields.asciidoc
+++ b/filebeat/docs/fields.asciidoc
@@ -18,6 +18,7 @@ grouped in the following categories:
 * <<exported-fields-cloud>>
 * <<exported-fields-docker-processor>>
 * <<exported-fields-icinga>>
+* <<exported-fields-kafka>>
 * <<exported-fields-kubernetes-processor>>
 * <<exported-fields-log>>
 * <<exported-fields-mysql>>
@@ -749,6 +750,95 @@ type: text
 The logged message.
 
 
+[[exported-fields-kafka]]
+== Kafka fields
+
+Kafka module
+
+
+
+[float]
+== kafka fields
+
+
+
+
+[float]
+== log fields
+
+Kafka log lines.
+
+
+
+[float]
+=== `kafka.log.timestamp`
+
+The timestamp from the log line.
+
+
+[float]
+=== `kafka.log.level`
+
+example: WARN
+
+The log level.
+
+
+[float]
+=== `kafka.log.message`
+
+type: text
+
+The logged message.
+
+
+[float]
+=== `kafka.log.component`
+
+type: keyword
+
+Component the log is coming from.
+
+
+[float]
+=== `kafka.log.class`
+
+type: text
+
+Java class the log is coming from.
+
+
+[float]
+== trace fields
+
+Trace in the log line.
+
+
+
+[float]
+=== `kafka.log.trace.class`
+
+type: keyword
+
+Java class the trace is coming from.
+
+
+[float]
+=== `kafka.log.trace.message`
+
+type: text
+
+Message part of the trace.
+
+
+[float]
+=== `kafka.log.trace.full`
+
+type: text
+
+The full trace in the log line.
+
+
 [[exported-fields-kubernetes-processor]]
 == Kubernetes fields
 
diff --git a/filebeat/docs/images/filebeat-kafka-logs-overview.png b/filebeat/docs/images/filebeat-kafka-logs-overview.png
new file mode 100644
index 000000000000..b04697a4b338
Binary files /dev/null and b/filebeat/docs/images/filebeat-kafka-logs-overview.png differ
diff --git a/filebeat/docs/modules/kafka.asciidoc b/filebeat/docs/modules/kafka.asciidoc
new file mode 100644
index 000000000000..36a13dc8ae53
--- /dev/null
+++ b/filebeat/docs/modules/kafka.asciidoc
@@ -0,0 +1,37 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[filebeat-module-kafka]]
+== Kafka module
+
+This module collects and parses the logs created by https://kafka.apache.org/[Kafka].
+
+[float]
+=== Compatibility
+
+The Kafka module was tested with logs from versions 2.11.
+
+[float]
+=== Dashboard
+
+This module comes with a sample dashboard to see Kafka logs and stacktraces.
+
+image::./images/filebeat-kafka-logs-overview.png[]
+
+[float]
+=== Logs fileset settings
+
+[float]
+==== var.paths
+
+An array of paths where to look for the log files. If left empty, Filebeat
+will choose the paths depending on your operating systems.
+
+
+[float]
+=== Fields
+
+For a description of each field in the metricset, see the
+<<exported-fields-kafka,exported fields>> section.
+
diff --git a/filebeat/docs/modules_list.asciidoc b/filebeat/docs/modules_list.asciidoc
index 1dd629002500..52dca849789d 100644
--- a/filebeat/docs/modules_list.asciidoc
+++ b/filebeat/docs/modules_list.asciidoc
@@ -6,6 +6,7 @@ This file is generated! See scripts/docs_collector.py
   * <<filebeat-module-apache2>>
   * <<filebeat-module-auditd>>
   * <<filebeat-module-icinga>>
+  * <<filebeat-module-kafka>>
   * <<filebeat-module-mysql>>
   * <<filebeat-module-nginx>>
   * <<filebeat-module-postgresql>>
@@ -19,6 +20,7 @@ include::modules-overview.asciidoc[]
 include::modules/apache2.asciidoc[]
 include::modules/auditd.asciidoc[]
 include::modules/icinga.asciidoc[]
+include::modules/kafka.asciidoc[]
 include::modules/mysql.asciidoc[]
 include::modules/nginx.asciidoc[]
 include::modules/postgresql.asciidoc[]
diff --git a/filebeat/filebeat.reference.yml b/filebeat/filebeat.reference.yml
index dcd7d03653be..d4566d31ccbc 100644
--- a/filebeat/filebeat.reference.yml
+++ b/filebeat/filebeat.reference.yml
@@ -114,6 +114,21 @@ filebeat.modules:
     # can be added under this section.
     #prospector:
 
+#-------------------------------- Kafka Module -------------------------------
+- module: kafka
+  # All logs
+  log:
+    enabled: true
+
+    # Set custom paths for Kafka. If left empty,
+    # Filebeat will look under /opt.
+    #var.kafka_home:
+
+    # Set custom paths for the log files. If left empty,
+    # Filebeat will choose the paths depending on your OS.
+    #var.paths:
+
+
 #-------------------------------- MySQL Module -------------------------------
 #- module: mysql
   # Error logs
diff --git a/filebeat/module/kafka/_meta/config.yml b/filebeat/module/kafka/_meta/config.yml
new file mode 100644
index 000000000000..fe568e8e5621
--- /dev/null
+++ b/filebeat/module/kafka/_meta/config.yml
@@ -0,0 +1,13 @@
+- module: kafka
+  # All logs
+  log:
+    enabled: true
+
+    # Set custom paths for Kafka. If left empty,
+    # Filebeat will look under /opt.
+    #var.kafka_home:
+
+    # Set custom paths for the log files. If left empty,
+    # Filebeat will choose the paths depending on your OS.
+    #var.paths:
+
diff --git a/filebeat/module/kafka/_meta/docs.asciidoc b/filebeat/module/kafka/_meta/docs.asciidoc
new file mode 100644
index 000000000000..fc2ea722bb57
--- /dev/null
+++ b/filebeat/module/kafka/_meta/docs.asciidoc
@@ -0,0 +1,24 @@
+== Kafka module
+
+This module collects and parses the logs created by https://kafka.apache.org/[Kafka].
+
+[float]
+=== Compatibility
+
+The Kafka module was tested with logs from versions 2.11.
+
+[float]
+=== Dashboard
+
+This module comes with a sample dashboard to see Kafka logs and stacktraces.
+
+image::./images/filebeat-kafka-logs-overview.png[]
+
+[float]
+=== Logs fileset settings
+
+[float]
+==== var.paths
+
+An array of paths where to look for the log files. If left empty, Filebeat
+will choose the paths depending on your operating systems.
diff --git a/filebeat/module/kafka/_meta/fields.yml b/filebeat/module/kafka/_meta/fields.yml
new file mode 100644
index 000000000000..5469c535e713
--- /dev/null
+++ b/filebeat/module/kafka/_meta/fields.yml
@@ -0,0 +1,9 @@
+- key: kafka
+  title: "Kafka"
+  description: >
+    Kafka module
+  fields:
+    - name: kafka
+      type: group
+      description: >
+      fields:
diff --git a/filebeat/module/kafka/_meta/kibana/default/dashboard/Filebeat-Kafka-overview.json b/filebeat/module/kafka/_meta/kibana/default/dashboard/Filebeat-Kafka-overview.json
new file mode 100644
index 000000000000..7b5d03e48eed
--- /dev/null
+++ b/filebeat/module/kafka/_meta/kibana/default/dashboard/Filebeat-Kafka-overview.json
@@ -0,0 +1,85 @@
+{
+  "objects": [
+    {
+      "attributes": {
+        "description": "",
+        "kibanaSavedObjectMeta": {
+          "searchSourceJSON": "{\"filter\":[]}"
+        },
+        "savedSearchId": "Kafka stacktraces",
+        "title": "Number of Kafka stracktraces by class",
+        "uiStateJSON": "{}",
+        "version": 1,
+        "visState": "{\"title\":\"Number of Kafka stracktraces by class\",\"type\":\"histogram\",\"params\":{\"type\":\"histogram\",\"grid\":{\"categoryLines\":false,\"style\":{\"color\":\"#eee\"}},\"categoryAxes\":[{\"id\":\"CategoryAxis-1\",\"type\":\"category\",\"position\":\"bottom\",\"show\":true,\"style\":{},\"scale\":{\"type\":\"linear\"},\"labels\":{\"show\":true,\"truncate\":100},\"title\":{\"text\":\"@timestamp per 30 minutes\"}}],\"valueAxes\":[{\"id\":\"ValueAxis-1\",\"name\":\"LeftAxis-1\",\"type\":\"value\",\"position\":\"left\",\"show\":true,\"style\":{},\"scale\":{\"type\":\"linear\",\"mode\":\"normal\"},\"labels\":{\"show\":true,\"rotate\":0,\"filter\":false,\"truncate\":100},\"title\":{\"text\":\"Count\"}}],\"seriesParams\":[{\"show\":\"true\",\"type\":\"histogram\",\"mode\":\"stacked\",\"data\":{\"label\":\"Count\",\"id\":\"1\"},\"valueAxis\":\"ValueAxis-1\",\"drawLinesBetweenPoints\":true,\"showCircles\":true}],\"addTooltip\":true,\"addLegend\":true,\"legendPosition\":\"right\",\"times\":[],\"addTimeMarker\":false},\"aggs\":[{\"id\":\"1\",\"enabled\":true,\"type\":\"count\",\"schema\":\"metric\",\"params\":{}},{\"id\":\"2\",\"enabled\":true,\"type\":\"date_histogram\",\"schema\":\"segment\",\"params\":{\"field\":\"@timestamp\",\"interval\":\"auto\",\"customInterval\":\"2h\",\"min_doc_count\":1,\"extended_bounds\":{}}},{\"id\":\"3\",\"enabled\":true,\"type\":\"terms\",\"schema\":\"group\",\"params\":{\"field\":\"kafka.log.trace.class\",\"size\":10,\"order\":\"desc\",\"orderBy\":\"1\"}}]}"
+      },
+      "id": "Number of Kafka stracktraces by class",
+      "type": "visualization",
+      "version": 1
+    },
+    {
+      "attributes": {
+        "columns": [
+          "kafka.log.class",
+          "kafka.log.trace.class",
+          "kafka.log.trace.full"
+        ],
+        "description": "",
+        "hits": 0,
+        "kibanaSavedObjectMeta": {
+          "searchSourceJSON": "{\"index\":\"filebeat-*\",\"highlightAll\":true,\"version\":true,\"query\":{\"query\":\"_exists_:kafka.log.trace.class\",\"language\":\"lucene\"},\"filter\":[]}"
+        },
+        "sort": [
+          "@timestamp",
+          "desc"
+        ],
+        "title": "Kafka stacktraces",
+        "version": 1
+      },
+      "id": "Kafka stacktraces",
+      "type": "search",
+      "version": 1
+    },
+    {
+      "attributes": {
+        "columns": [
+          "kafka.log.level",
+          "kafka.log.component",
+          "kafka.log.message"
+        ],
+        "description": "",
+        "hits": 0,
+        "kibanaSavedObjectMeta": {
+          "searchSourceJSON": "{\"index\":\"filebeat-*\",\"highlightAll\":true,\"version\":true,\"query\":{\"query\":\"kafka.log.level:*\",\"language\":\"lucene\"},\"filter\":[]}"
+        },
+        "sort": [
+          "@timestamp",
+          "desc"
+        ],
+        "title": "All Kafka logs",
+        "version": 1
+      },
+      "id": "All Kafka logs",
+      "type": "search",
+      "version": 1
+    },
+    {
+      "attributes": {
+        "description": "",
+        "hits": 0,
+        "kibanaSavedObjectMeta": {
+          "searchSourceJSON": "{\"query\":{\"language\":\"lucene\",\"query\":\"\"},\"filter\":[],\"highlightAll\":true,\"version\":true}"
+        },
+        "optionsJSON": "{\"darkTheme\":false}",
+        "panelsJSON": "[{\"col\":1,\"id\":\"Number of Kafka stracktraces by class\",\"panelIndex\":1,\"row\":1,\"size_x\":6,\"size_y\":3,\"type\":\"visualization\"},{\"col\":7,\"columns\":[\"kafka.log.class\",\"kafka.log.trace.class\",\"kafka.log.trace.full\"],\"id\":\"Kafka stacktraces\",\"panelIndex\":2,\"row\":1,\"size_x\":6,\"size_y\":3,\"sort\":[\"@timestamp\",\"desc\"],\"type\":\"search\"},{\"size_x\":12,\"size_y\":3,\"panelIndex\":3,\"type\":\"search\",\"id\":\"All Kafka logs\",\"col\":1,\"row\":4,\"columns\":[\"kafka.log.level\",\"kafka.log.component\",\"kafka.log.message\"],\"sort\":[\"@timestamp\",\"desc\"]}]",
+        "timeRestore": false,
+        "title": "Kafka logs overview",
+        "uiStateJSON": "{}",
+        "version": 1
+      },
+      "id": "Kafka logs overview",
+      "type": "dashboard",
+      "version": 1
+    }
+  ],
+  "version": "6.0.0-beta1-SNAPSHOT"
+}
diff --git a/filebeat/module/kafka/log/_meta/fields.yml b/filebeat/module/kafka/log/_meta/fields.yml
new file mode 100644
index 000000000000..e9d234cb625e
--- /dev/null
+++ b/filebeat/module/kafka/log/_meta/fields.yml
@@ -0,0 +1,41 @@
+- name: log
+  type: group
+  description: >
+    Kafka log lines.
+  fields:
+    - name: timestamp
+      description: >
+        The timestamp from the log line.
+    - name: level
+      example: "WARN"
+      description: >
+        The log level.
+    - name: message
+      type: text
+      description: >
+        The logged message.
+    - name: component
+      type: keyword
+      description: >
+        Component the log is coming from.
+    - name: class
+      type: text
+      description: >
+        Java class the log is coming from.
+    - name: trace
+      type: group
+      description: >
+          Trace in the log line.
+      fields:
+        - name: class
+          type: keyword
+          description: >
+            Java class the trace is coming from.
+        - name: message
+          type: text
+          description: >
+              Message part of the trace.
+        - name: full
+          type: text
+          description: >
+              The full trace in the log line.
diff --git a/filebeat/module/kafka/log/config/log.yml b/filebeat/module/kafka/log/config/log.yml
new file mode 100644
index 000000000000..98c684ef52f6
--- /dev/null
+++ b/filebeat/module/kafka/log/config/log.yml
@@ -0,0 +1,10 @@
+type: log
+paths:
+{{ range $i, $path := .paths }}
+ - {{$path}}
+{{ end }}
+exclude_files: [".gz$"]
+multiline:
+  pattern: '^\['
+  negate: true
+  match: after
diff --git a/filebeat/module/kafka/log/ingest/pipeline.json b/filebeat/module/kafka/log/ingest/pipeline.json
new file mode 100644
index 000000000000..1c5149f5a511
--- /dev/null
+++ b/filebeat/module/kafka/log/ingest/pipeline.json
@@ -0,0 +1,66 @@
+{
+  "description": "Pipeline for parsing Kafka log messages",
+  "processors": [
+    {
+      "grok": {
+        "field": "message",
+        "trace_match": true,
+        "patterns": [
+          "(?m)%{TIMESTAMP_ISO8601:kafka.log.timestamp}. %{LOGLEVEL:kafka.log.level} +%{JAVALOGMESSAGE:kafka.log.message} \\(%{JAVACLASS:kafka.log.class}\\)$[ \\n]*(?'kafka.log.trace.full'.*)"
+        ]
+      }
+    },
+    {
+      "grok": {
+        "field": "kafka.log.message",
+        "patterns": [
+          "\\[%{DATA:kafka.log.component}\\][,:. ] +%{JAVALOGMESSAGE:kafka.log.message}"
+        ],
+        "on_failure": [
+          {
+            "set": {
+              "field": "kafka.log.component",
+              "value": "unknown"
+            }
+          }
+        ]
+      }
+    },
+    {
+      "grok": {
+        "field": "kafka.log.trace.full",
+        "ignore_missing": true,
+        "patterns": [
+          "%{JAVACLASS:kafka.log.trace.class}:\\s*%{JAVALOGMESSAGE:kafka.log.trace.message}"
+        ],
+        "on_failure": [
+          {
+            "remove": {
+              "field": "kafka.log.trace"
+            }
+          }
+        ]
+      }
+    },
+    {
+      "rename": {
+        "field": "@timestamp",
+        "target_field": "beat.read_time"
+      }
+    },
+    {
+      "date": {
+        "field": "kafka.log.timestamp",
+        "target_field": "@timestamp",
+        "formats": ["yyyy-MM-dd HH:mm:ss,SSS"]
+      }
+    },
+    {"remove": {"field": "kafka.log.timestamp" }}
+  ],
+  "on_failure" : [{
+    "set" : {
+      "field" : "error.log",
+      "value" : "{{ _ingest.on_failure_message }}"
+    }
+  }]
+}
diff --git a/filebeat/module/kafka/log/manifest.yml b/filebeat/module/kafka/log/manifest.yml
new file mode 100644
index 000000000000..fb7916d9536f
--- /dev/null
+++ b/filebeat/module/kafka/log/manifest.yml
@@ -0,0 +1,14 @@
+module_version: 1.0
+
+var:
+  - name: kafka_home
+    default: /opt/kafka*
+  - name: paths
+    default:
+      - "{{.kafka_home}}/logs/controller.log*"
+      - "{{.kafka_home}}/logs/server.log*"
+      - "{{.kafka_home}}/logs/state-change.log*"
+      - "{{.kafka_home}}/logs/kafka-*.log*"
+
+ingest_pipeline: ingest/pipeline.json
+prospector: config/log.yml
diff --git a/filebeat/module/kafka/log/test/controller.log b/filebeat/module/kafka/log/test/controller.log
new file mode 100644
index 000000000000..39cd4b970437
--- /dev/null
+++ b/filebeat/module/kafka/log/test/controller.log
@@ -0,0 +1,20 @@
+[2017-08-04 10:48:21,048] INFO [controller-event-thread]: Starting (kafka.controller.ControllerEventManager$ControllerEventThread)
+[2017-08-04 10:48:21,063] INFO [Controller 0]: 0 successfully elected as the controller (kafka.controller.KafkaController)
+[2017-08-04 10:48:21,064] INFO [Controller 0]: Broker 0 starting become controller state transition (kafka.controller.KafkaController)
+[2017-08-04 10:48:21,082] INFO [Controller 0]: Controller 0 incremented epoch to 1 (kafka.controller.KafkaController)
+[2017-08-04 10:48:21,085] DEBUG [Controller 0]: Registering IsrChangeNotificationListener (kafka.controller.KafkaController)
+[2017-08-04 10:48:21,154] INFO [Replica state machine on controller 0]: Started replica state machine with initial state -> Map() (kafka.controller.ReplicaStateMachine)
+[2017-08-04 10:48:21,156] INFO [Partition state machine on Controller 0]: Started partition state machine with initial state -> Map() (kafka.controller.PartitionStateMachine)
+[2017-08-04 10:48:21,157] INFO [Controller 0]: Broker 0 is ready to serve as the new controller with epoch 1 (kafka.controller.KafkaController)
+[2017-08-04 10:48:21,165] INFO [Partition state machine on Controller 0]: Invoking state change to OnlinePartition for partitions  (kafka.controller.PartitionStateMachine)
+[2017-08-04 11:44:22,588] DEBUG [Controller 0]: Live brokers:  (kafka.controller.KafkaController)
+[2017-08-04 11:44:25,094] INFO [controller-event-thread]: Shutting down (kafka.controller.ControllerEventManager$ControllerEventThread)
+[2017-08-04 11:44:25,095] INFO [controller-event-thread]: Stopped (kafka.controller.ControllerEventManager$ControllerEventThread)
+[2017-08-04 11:44:25,097] INFO [controller-event-thread]: Shutdown completed (kafka.controller.ControllerEventManager$ControllerEventThread)
+[2017-08-04 11:44:25,099] DEBUG [Controller 0]: Controller resigning, broker id 0 (kafka.controller.KafkaController)
+[2017-08-04 11:44:25,100] DEBUG [Controller 0]: De-registering IsrChangeNotificationListener (kafka.controller.KafkaController)
+[2017-08-04 11:44:25,105] INFO [Partition state machine on Controller 0]: Stopped partition state machine (kafka.controller.PartitionStateMachine)
+[2017-08-04 11:44:25,111] INFO [Replica state machine on controller 0]: Stopped replica state machine (kafka.controller.ReplicaStateMachine)
+[2017-08-04 11:44:25,112] INFO [Controller-0-to-broker-0-send-thread]: Shutting down (kafka.controller.RequestSendThread)
+[2017-08-04 11:44:25,112] INFO [Controller-0-to-broker-0-send-thread]: Stopped (kafka.controller.RequestSendThread)
+[2017-08-04 11:44:25,113] INFO [Controller-0-to-broker-0-send-thread]: Shutdown completed (kafka.controller.RequestSendThread)
diff --git a/filebeat/module/kafka/log/test/controller.log-expected.json b/filebeat/module/kafka/log/test/controller.log-expected.json
new file mode 100644
index 000000000000..b3453298e434
--- /dev/null
+++ b/filebeat/module/kafka/log/test/controller.log-expected.json
@@ -0,0 +1,961 @@
+[
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "wItR0l0Bed9fOYVUjezS",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.113Z",
+      "offset": 2695,
+      "kafka": {
+        "log": {
+          "component": "Controller-0-to-broker-0-send-thread",
+          "level": "INFO",
+          "message": "Shutdown completed",
+          "class": "kafka.controller.RequestSendThread"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.376Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 11:44:25,113] INFO [Controller-0-to-broker-0-send-thread]: Shutdown completed (kafka.controller.RequestSendThread)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.113Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065113
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "notR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.112Z",
+      "offset": 2452,
+      "kafka": {
+        "log": {
+          "component": "Controller-0-to-broker-0-send-thread",
+          "level": "INFO",
+          "message": "Shutting down",
+          "class": "kafka.controller.RequestSendThread"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.374Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 11:44:25,112] INFO [Controller-0-to-broker-0-send-thread]: Shutting down (kafka.controller.RequestSendThread)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.112Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065112
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "n4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.112Z",
+      "offset": 2568,
+      "kafka": {
+        "log": {
+          "component": "Controller-0-to-broker-0-send-thread",
+          "level": "INFO",
+          "message": "Stopped",
+          "class": "kafka.controller.RequestSendThread"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.376Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 11:44:25,112] INFO [Controller-0-to-broker-0-send-thread]: Stopped (kafka.controller.RequestSendThread)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.112Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065112
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "nYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.111Z",
+      "offset": 2330,
+      "kafka": {
+        "log": {
+          "component": "Replica state machine on controller 0",
+          "level": "INFO",
+          "message": "Stopped replica state machine",
+          "class": "kafka.controller.ReplicaStateMachine"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.373Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 11:44:25,111] INFO [Replica state machine on controller 0]: Stopped replica state machine (kafka.controller.ReplicaStateMachine)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.111Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065111
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "nItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.105Z",
+      "offset": 2189,
+      "kafka": {
+        "log": {
+          "component": "Partition state machine on Controller 0",
+          "level": "INFO",
+          "message": "Stopped partition state machine",
+          "class": "kafka.controller.PartitionStateMachine"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.372Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 11:44:25,105] INFO [Partition state machine on Controller 0]: Stopped partition state machine (kafka.controller.PartitionStateMachine)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.105Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065105
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "m4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.100Z",
+      "offset": 2042,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "DEBUG",
+          "message": "De-registering IsrChangeNotificationListener",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.372Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 11:44:25,100] DEBUG [Controller 0]: De-registering IsrChangeNotificationListener (kafka.controller.KafkaController)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.100Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065100
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "motR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.099Z",
+      "offset": 1914,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "DEBUG",
+          "message": "Controller resigning, broker id 0",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.372Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 11:44:25,099] DEBUG [Controller 0]: Controller resigning, broker id 0 (kafka.controller.KafkaController)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.099Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065099
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "mYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.097Z",
+      "offset": 1797,
+      "kafka": {
+        "log": {
+          "component": "controller-event-thread",
+          "level": "INFO",
+          "message": "Shutdown completed",
+          "class": "kafka.controller.ControllerEventManager$ControllerEventThread"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.370Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 11:44:25,097] INFO [controller-event-thread]: Shutdown completed (kafka.controller.ControllerEventManager$ControllerEventThread)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.097Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065097
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "mItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.095Z",
+      "offset": 1656,
+      "kafka": {
+        "log": {
+          "component": "controller-event-thread",
+          "level": "INFO",
+          "message": "Stopped",
+          "class": "kafka.controller.ControllerEventManager$ControllerEventThread"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.369Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 11:44:25,095] INFO [controller-event-thread]: Stopped (kafka.controller.ControllerEventManager$ControllerEventThread)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.095Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065095
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "l4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:25.094Z",
+      "offset": 1526,
+      "kafka": {
+        "log": {
+          "component": "controller-event-thread",
+          "level": "INFO",
+          "message": "Shutting down",
+          "class": "kafka.controller.ControllerEventManager$ControllerEventThread"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.367Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 11:44:25,094] INFO [controller-event-thread]: Shutting down (kafka.controller.ControllerEventManager$ControllerEventThread)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:25.094Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847065094
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "lotR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T11:44:22.588Z",
+      "offset": 1390,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "DEBUG",
+          "message": "Live brokers: ",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.365Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 11:44:22,588] DEBUG [Controller 0]: Live brokers:  (kafka.controller.KafkaController)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T11:44:22.588Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501847062588
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "lYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.165Z",
+      "offset": 1292,
+      "kafka": {
+        "log": {
+          "component": "Partition state machine on Controller 0",
+          "level": "INFO",
+          "message": "Invoking state change to OnlinePartition for partitions ",
+          "class": "kafka.controller.PartitionStateMachine"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.364Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 10:48:21,165] INFO [Partition state machine on Controller 0]: Invoking state change to OnlinePartition for partitions  (kafka.controller.PartitionStateMachine)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.165Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701165
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "lItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.157Z",
+      "offset": 1120,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "INFO",
+          "message": "Broker 0 is ready to serve as the new controller with epoch 1",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.363Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:21,157] INFO [Controller 0]: Broker 0 is ready to serve as the new controller with epoch 1 (kafka.controller.KafkaController)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.157Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701157
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "k4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.156Z",
+      "offset": 976,
+      "kafka": {
+        "log": {
+          "component": "Partition state machine on Controller 0",
+          "level": "INFO",
+          "message": "Started partition state machine with initial state -> Map()",
+          "class": "kafka.controller.PartitionStateMachine"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.362Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 10:48:21,156] INFO [Partition state machine on Controller 0]: Started partition state machine with initial state -> Map() (kafka.controller.PartitionStateMachine)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.156Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701156
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "kotR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.154Z",
+      "offset": 801,
+      "kafka": {
+        "log": {
+          "component": "Replica state machine on controller 0",
+          "level": "INFO",
+          "message": "Started replica state machine with initial state -> Map()",
+          "class": "kafka.controller.ReplicaStateMachine"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.361Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 10:48:21,154] INFO [Replica state machine on controller 0]: Started replica state machine with initial state -> Map() (kafka.controller.ReplicaStateMachine)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.154Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701154
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "kYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.085Z",
+      "offset": 632,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "DEBUG",
+          "message": "Registering IsrChangeNotificationListener",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.360Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:21,085] DEBUG [Controller 0]: Registering IsrChangeNotificationListener (kafka.controller.KafkaController)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.085Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701085
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "kItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.082Z",
+      "offset": 507,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "INFO",
+          "message": "Controller 0 incremented epoch to 1",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.358Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 10:48:21,082] INFO [Controller 0]: Controller 0 incremented epoch to 1 (kafka.controller.KafkaController)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.082Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701082
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "j4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.064Z",
+      "offset": 389,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "INFO",
+          "message": "Broker 0 starting become controller state transition",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.358Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "message": "[2017-08-04 10:48:21,064] INFO [Controller 0]: Broker 0 starting become controller state transition (kafka.controller.KafkaController)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.064Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701064
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "jotR0l0Bed9fOYVUfexP",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.063Z",
+      "offset": 254,
+      "kafka": {
+        "log": {
+          "component": "Controller 0",
+          "level": "INFO",
+          "message": "0 successfully elected as the controller",
+          "class": "kafka.controller.KafkaController"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.358Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:21,063] INFO [Controller 0]: 0 successfully elected as the controller (kafka.controller.KafkaController)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.063Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701063
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "jYtR0l0Bed9fOYVUfOzU",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.048Z",
+      "offset": 131,
+      "kafka": {
+        "log": {
+          "component": "controller-event-thread",
+          "level": "INFO",
+          "message": "Starting",
+          "class": "kafka.controller.ControllerEventManager$ControllerEventThread"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.358Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:21,048] INFO [controller-event-thread]: Starting (kafka.controller.ControllerEventManager$ControllerEventThread)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.048Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/controller.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701048
+    ]
+  }
+]
diff --git a/filebeat/module/kafka/log/test/server.log b/filebeat/module/kafka/log/test/server.log
new file mode 100644
index 000000000000..8ebf66523d9f
--- /dev/null
+++ b/filebeat/module/kafka/log/test/server.log
@@ -0,0 +1,20 @@
+[2017-08-04 10:48:20,377] INFO starting (kafka.server.KafkaServer)
+[2017-08-04 10:48:20,379] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)
+[2017-08-04 10:48:20,400] INFO Client environment:java.io.tmpdir=/tmp (org.apache.zookeeper.ZooKeeper)
+[2017-08-04 10:48:20,400] INFO Client environment:java.compiler=<NA> (org.apache.zookeeper.ZooKeeper)
+[2017-08-04 10:48:20,401] INFO Initiating client connection, connectString=localhost:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@5ffead27 (org.apache.zookeeper.ZooKeeper)
+[2017-08-04 10:48:20,413] INFO Waiting for keeper state SyncConnected (org.I0Itec.zkclient.ZkClient)
+[2017-08-04 10:48:20,415] INFO Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
+[2017-08-04 10:48:20,420] INFO Socket connection established to localhost/0:0:0:0:0:0:0:1:2181, initiating session (org.apache.zookeeper.ClientCnxn)
+[2017-08-04 10:48:20,457] INFO Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x15dabf8d4140000, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
+[2017-08-04 10:48:20,458] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)
+[2017-08-04 10:48:20,748] WARN No meta.properties file under dir /tmp/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
+[2017-08-04 10:48:20,800] INFO [ThrottledRequestReaper-Fetch]: Starting (kafka.server.ClientQuotaManager$ThrottledRequestReaper)
+[2017-08-04 10:48:20,866] INFO Log directory '/tmp/kafka-logs' not found, creating it. (kafka.log.LogManager)
+[2017-08-04 10:48:20,873] INFO Loading logs. (kafka.log.LogManager)
+[2017-08-04 10:48:21,062] INFO [ExpirationReaper-0-Heartbeat]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2017-08-04 10:48:21,063] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
+[2017-08-04 10:48:21,095] INFO [Group Metadata Manager on Broker 0]: Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
+[2017-08-04 10:48:21,127] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:0,blockEndProducerId:999) by writing to Zk with path version 1 (kafka.coordinator.transaction.ProducerIdManager)
+[2017-08-04 10:48:21,162] INFO [Transaction Coordinator 0]: Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
+[2017-08-04 10:48:21,167] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
diff --git a/filebeat/module/kafka/log/test/server.log-expected.json b/filebeat/module/kafka/log/test/server.log-expected.json
new file mode 100644
index 000000000000..713942361b5a
--- /dev/null
+++ b/filebeat/module/kafka/log/test/server.log-expected.json
@@ -0,0 +1,961 @@
+[
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "wYtR0l0Bed9fOYVUjuwB",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.167Z",
+      "offset": 2641,
+      "kafka": {
+        "log": {
+          "component": "Transaction Marker Channel Manager 0",
+          "level": "INFO",
+          "message": "Starting",
+          "class": "kafka.coordinator.transaction.TransactionMarkerChannelManager"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.385Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:21,167] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.167Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701167
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "sotR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.162Z",
+      "offset": 2497,
+      "kafka": {
+        "log": {
+          "component": "Transaction Coordinator 0",
+          "level": "INFO",
+          "message": "Starting up.",
+          "class": "kafka.coordinator.transaction.TransactionCoordinator"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.384Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:21,162] INFO [Transaction Coordinator 0]: Starting up. (kafka.coordinator.transaction.TransactionCoordinator)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.162Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701162
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "sYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.127Z",
+      "offset": 2369,
+      "kafka": {
+        "log": {
+          "component": "ProducerId Manager 0",
+          "level": "INFO",
+          "message": "Acquired new producerId block (brokerId:0,blockStartProducerId:0,blockEndProducerId:999) by writing to Zk with path version 1",
+          "class": "kafka.coordinator.transaction.ProducerIdManager"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.384Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:21,127] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:0,blockEndProducerId:999) by writing to Zk with path version 1 (kafka.coordinator.transaction.ProducerIdManager)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.127Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701127
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "sItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.095Z",
+      "offset": 2138,
+      "kafka": {
+        "log": {
+          "component": "Group Metadata Manager on Broker 0",
+          "level": "INFO",
+          "message": "Removed 0 expired offsets in 1 milliseconds.",
+          "class": "kafka.coordinator.group.GroupMetadataManager"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.384Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:21,095] INFO [Group Metadata Manager on Broker 0]: Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.095Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701095
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "r4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.063Z",
+      "offset": 1977,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Result of znode creation is: OK",
+          "class": "kafka.utils.ZKCheckedEphemeral"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.384Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:21,063] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.063Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701063
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "rotR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.062Z",
+      "offset": 1881,
+      "kafka": {
+        "log": {
+          "component": "ExpirationReaper-0-Heartbeat",
+          "level": "INFO",
+          "message": "Starting",
+          "class": "kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.381Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:21,062] INFO [ExpirationReaper-0-Heartbeat]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.062Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701062
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "rYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.873Z",
+      "offset": 1745,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Loading logs.",
+          "class": "kafka.log.LogManager"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.381Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,873] INFO Loading logs. (kafka.log.LogManager)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.873Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700873
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "rItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.866Z",
+      "offset": 1677,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Log directory '/tmp/kafka-logs' not found, creating it.",
+          "class": "kafka.log.LogManager"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.381Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:20,866] INFO Log directory '/tmp/kafka-logs' not found, creating it. (kafka.log.LogManager)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.866Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700866
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "q4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.800Z",
+      "offset": 1567,
+      "kafka": {
+        "log": {
+          "component": "ThrottledRequestReaper-Fetch",
+          "level": "INFO",
+          "message": "Starting",
+          "class": "kafka.server.ClientQuotaManager$ThrottledRequestReaper"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.380Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,800] INFO [ThrottledRequestReaper-Fetch]: Starting (kafka.server.ClientQuotaManager$ThrottledRequestReaper)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.800Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700800
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "qotR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.748Z",
+      "offset": 1438,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "WARN",
+          "message": "No meta.properties file under dir /tmp/kafka-logs/meta.properties",
+          "class": "kafka.server.BrokerMetadataCheckpoint"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.380Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,748] WARN No meta.properties file under dir /tmp/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.748Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700748
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "qYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.458Z",
+      "offset": 1301,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "zookeeper state changed (SyncConnected)",
+          "class": "org.I0Itec.zkclient.ZkClient"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.379Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:20,458] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.458Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700458
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "qItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.457Z",
+      "offset": 1199,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x15dabf8d4140000, negotiated timeout = 6000",
+          "class": "org.apache.zookeeper.ClientCnxn"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.379Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,457] INFO Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x15dabf8d4140000, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.457Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700457
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "p4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.420Z",
+      "offset": 1004,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Socket connection established to localhost/0:0:0:0:0:0:0:1:2181, initiating session",
+          "class": "org.apache.zookeeper.ClientCnxn"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.379Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,420] INFO Socket connection established to localhost/0:0:0:0:0:0:0:1:2181, initiating session (org.apache.zookeeper.ClientCnxn)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.420Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700420
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "potR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.415Z",
+      "offset": 855,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error)",
+          "class": "org.apache.zookeeper.ClientCnxn"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.379Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,415] INFO Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.415Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700415
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "pYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.413Z",
+      "offset": 662,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Waiting for keeper state SyncConnected",
+          "class": "org.I0Itec.zkclient.ZkClient"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.378Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,413] INFO Waiting for keeper state SyncConnected (org.I0Itec.zkclient.ZkClient)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.413Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700413
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "pItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.401Z",
+      "offset": 561,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Initiating client connection, connectString=localhost:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@5ffead27",
+          "class": "org.apache.zookeeper.ZooKeeper"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.378Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,401] INFO Initiating client connection, connectString=localhost:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@5ffead27 (org.apache.zookeeper.ZooKeeper)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.401Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700401
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "o4tR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.400Z",
+      "offset": 372,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Client environment:java.compiler=<NA>",
+          "class": "org.apache.zookeeper.ZooKeeper"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.359Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,400] INFO Client environment:java.compiler=<NA> (org.apache.zookeeper.ZooKeeper)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.400Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700400
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "ootR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.400Z",
+      "offset": 270,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Client environment:java.io.tmpdir=/tmp",
+          "class": "org.apache.zookeeper.ZooKeeper"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.359Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,400] INFO Client environment:java.io.tmpdir=/tmp (org.apache.zookeeper.ZooKeeper)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.400Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700400
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "oYtR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.379Z",
+      "offset": 167,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "Connecting to zookeeper on localhost:2181",
+          "class": "kafka.server.KafkaServer"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.359Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "message": "[2017-08-04 10:48:20,379] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.379Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700379
+    ]
+  },
+  
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "oItR0l0Bed9fOYVUfeyz",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:20.377Z",
+      "offset": 67,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "INFO",
+          "message": "starting",
+          "class": "kafka.server.KafkaServer"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.359Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      },
+      "message": "[2017-08-04 10:48:20,377] INFO starting (kafka.server.KafkaServer)"
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:20.377Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/server.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843700377
+    ]
+  }
+]
diff --git a/filebeat/module/kafka/log/test/state-change.log b/filebeat/module/kafka/log/test/state-change.log
new file mode 100644
index 000000000000..675a51dff725
--- /dev/null
+++ b/filebeat/module/kafka/log/test/state-change.log
@@ -0,0 +1 @@
+[2017-08-04 10:48:21,428] TRACE Controller 0 epoch 1 received response {error_code=0} for a request sent to broker baldur:9092 (id: 0 rack: null) (state.change.logger)
diff --git a/filebeat/module/kafka/log/test/state-change.log-expected.json b/filebeat/module/kafka/log/test/state-change.log-expected.json
new file mode 100644
index 000000000000..9d02b4d88696
--- /dev/null
+++ b/filebeat/module/kafka/log/test/state-change.log-expected.json
@@ -0,0 +1,49 @@
+[
+  {
+    "_index": "filebeat-7.0.0-alpha1-2017.08.11",
+    "_type": "doc",
+    "_id": "v4tR0l0Bed9fOYVUjey6",
+    "_version": 1,
+    "_score": null,
+    "_source": {
+      "@timestamp": "2017-08-04T10:48:21.428Z",
+      "offset": 168,
+      "kafka": {
+        "log": {
+          "component": "unknown",
+          "level": "TRACE",
+          "message": "Controller 0 epoch 1 received response {error_code=0} for a request sent to broker baldur:9092 (id: 0 rack: null)",
+          "class": "state.change.logger"
+        }
+      },
+      "beat": {
+        "hostname": "baldur",
+        "read_time": "2017-08-11T17:22:21.359Z",
+        "name": "baldur",
+        "version": "7.0.0-alpha1"
+      },
+      "prospector": {
+        "type": "log"
+      },
+      "source": "/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/state-change.log",
+      "message": "[2017-08-04 10:48:21,428] TRACE Controller 0 epoch 1 received response {error_code=0} for a request sent to broker baldur:9092 (id: 0 rack: null) (state.change.logger)",
+      "fileset": {
+        "module": "kafka",
+        "name": "log"
+      }
+    },
+    "fields": {
+      "@timestamp": [
+        "2017-08-04T10:48:21.428Z"
+      ]
+    },
+    "highlight": {
+      "source": [
+        "@kibana-highlighted-field@/home/n/go/src/github.com/elastic/beats/filebeat/module/kafka/log/test/state-change.log@/kibana-highlighted-field@"
+      ]
+    },
+    "sort": [
+      1501843701428
+    ]
+  }
+]
diff --git a/filebeat/modules.d/kafka.yml.disabled b/filebeat/modules.d/kafka.yml.disabled
new file mode 100644
index 000000000000..fe568e8e5621
--- /dev/null
+++ b/filebeat/modules.d/kafka.yml.disabled
@@ -0,0 +1,13 @@
+- module: kafka
+  # All logs
+  log:
+    enabled: true
+
+    # Set custom paths for Kafka. If left empty,
+    # Filebeat will look under /opt.
+    #var.kafka_home:
+
+    # Set custom paths for the log files. If left empty,
+    # Filebeat will choose the paths depending on your OS.
+    #var.paths:
+