diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index a8a106e6535..746336157e0 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -152,6 +152,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 - Added default and TCP parsing formats to HAproxy module {issue}8311[8311] {pull}8637[8637]
 - Add Suricata IDS/IDP/NSM module. {issue}8153[8153] {pull}8693[8693]
 - Allow to force CRI format parsing for better performance {pull}8424[8424]
+- Support for Kafka 2.0.0 {pull}8853[8853]
 
 *Heartbeat*
 
diff --git a/filebeat/docs/modules/kafka.asciidoc b/filebeat/docs/modules/kafka.asciidoc
index 278e2141084..54d7b69eb43 100644
--- a/filebeat/docs/modules/kafka.asciidoc
+++ b/filebeat/docs/modules/kafka.asciidoc
@@ -16,7 +16,7 @@ include::../include/what-happens.asciidoc[]
 [float]
 === Compatibility
 
-The +{modulename}+ module was tested with logs from versions 0.9 and 1.11.
+The +{modulename}+ module was tested with logs from versions 0.9, 1.1.0 and 2.0.0.
 
 include::../include/running-modules.asciidoc[]
 
diff --git a/filebeat/module/kafka/_meta/docs.asciidoc b/filebeat/module/kafka/_meta/docs.asciidoc
index 0d2a39292c0..25625374898 100644
--- a/filebeat/module/kafka/_meta/docs.asciidoc
+++ b/filebeat/module/kafka/_meta/docs.asciidoc
@@ -11,7 +11,7 @@ include::../include/what-happens.asciidoc[]
 [float]
 === Compatibility
 
-The +{modulename}+ module was tested with logs from versions 0.9 and 1.11.
+The +{modulename}+ module was tested with logs from versions 0.9, 1.1.0 and 2.0.0.
 
 include::../include/running-modules.asciidoc[]
 
diff --git a/filebeat/module/kafka/log/test/controller-2.0.0.log b/filebeat/module/kafka/log/test/controller-2.0.0.log
new file mode 100644
index 00000000000..52a32f099b9
--- /dev/null
+++ b/filebeat/module/kafka/log/test/controller-2.0.0.log
@@ -0,0 +1,23 @@
+[2018-10-31 15:03:32,474] TRACE [Controller id=10] Checking need to trigger auto leader balancing (kafka.controller.KafkaController)
+[2018-10-31 15:03:32,474] DEBUG [Controller id=10] Preferred replicas by broker Map(20 -> Map(__consumer_offsets-22 -> Vector(20), __consumer_offsets-4 -> Vector(20), __consumer_offsets-7 -> Vector(20), __consumer_offsets-46 -> Vector(20), __consumer_offsets-25 -> Vector(20), __consumer_offsets-49 -> Vector(20), __consumer_offsets-16 -> Vector(20), test-0 -> Vector(20, 30, 10), __consumer_offsets-28 -> Vector(20), __consumer_offsets-31 -> Vector(20), test-2-2 -> Vector(20, 30), __consumer_offsets-37 -> Vector(20), filebeat-system-0 -> Vector(20), test-3-3 -> Vector(20, 30), __consumer_offsets-19 -> Vector(20), __consumer_offsets-13 -> Vector(20), __consumer_offsets-43 -> Vector(20), __consumer_offsets-1 -> Vector(20), __consumer_offsets-34 -> Vector(20), __consumer_offsets-10 -> Vector(20), test-3 -> Vector(20, 10, 30), __consumer_offsets-40 -> Vector(20)), 10 -> Map(__consumer_offsets-30 -> Vector(10), __consumer_offsets-21 -> Vector(10), __consumer_offsets-27 -> Vector(10), __consumer_offsets-9 -> Vector(10), __consumer_offsets-33 -> Vector(10), __consumer_offsets-36 -> Vector(10), __consumer_offsets-42 -> Vector(10), __consumer_offsets-3 -> Vector(10), __consumer_offsets-18 -> Vector(10), test-5 -> Vector(10, 30, 20), __consumer_offsets-15 -> Vector(10), __consumer_offsets-24 -> Vector(10), test-3-1 -> Vector(10, 20), __consumer_offsets-48 -> Vector(10), filebeat-kafka-0 -> Vector(10), __consumer_offsets-6 -> Vector(10), test-2-1 -> Vector(10, 20), test-3-2 -> Vector(10, 30), __consumer_offsets-0 -> Vector(10), __consumer_offsets-39 -> Vector(10), __consumer_offsets-12 -> Vector(10), __consumer_offsets-45 -> Vector(10), test-2 -> Vector(10, 20, 30)), 30 -> Map(__consumer_offsets-8 -> Vector(30), __consumer_offsets-35 -> Vector(30), __consumer_offsets-41 -> Vector(30), __consumer_offsets-23 -> Vector(30), __consumer_offsets-47 -> Vector(30), metricbeat-0 -> Vector(30), test-3-0 -> Vector(30, 10), metricbeat-kafka-0 -> Vector(30), filebeat-0 -> Vector(30), test-2-0 -> Vector(30, 10), __consumer_offsets-38 -> Vector(30), __consumer_offsets-17 -> Vector(30), test-1 -> Vector(30, 10, 20), test-2-3 -> Vector(30, 20), __consumer_offsets-11 -> Vector(30), __consumer_offsets-2 -> Vector(30), __consumer_offsets-14 -> Vector(30), test-4 -> Vector(30, 20, 10), metricbeat-system-0 -> Vector(30), __consumer_offsets-20 -> Vector(30), __consumer_offsets-44 -> Vector(30), __consumer_offsets-5 -> Vector(30), __consumer_offsets-26 -> Vector(30), __consumer_offsets-29 -> Vector(30), __consumer_offsets-32 -> Vector(30))) (kafka.controller.KafkaController)
+[2018-10-31 15:03:32,474] DEBUG [Controller id=10] Topics not in preferred replica for broker 20 Map() (kafka.controller.KafkaController)
+[2018-10-31 15:03:32,475] TRACE [Controller id=10] Leader imbalance ratio for broker 20 is 0.0 (kafka.controller.KafkaController)
+[2018-10-31 15:03:32,475] DEBUG [Controller id=10] Topics not in preferred replica for broker 10 Map(test-3-1 -> Vector(10, 20)) (kafka.controller.KafkaController)
+[2018-10-31 15:03:32,475] TRACE [Controller id=10] Leader imbalance ratio for broker 10 is 0.043478260869565216 (kafka.controller.KafkaController)
+[2018-10-31 15:03:32,475] DEBUG [Controller id=10] Topics not in preferred replica for broker 30 Map() (kafka.controller.KafkaController)
+[2018-10-31 15:03:32,475] TRACE [Controller id=10] Leader imbalance ratio for broker 30 is 0.0 (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] TRACE [Controller id=10] Checking need to trigger auto leader balancing (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] DEBUG [Controller id=10] Preferred replicas by broker Map(20 -> Map(__consumer_offsets-22 -> Vector(20), __consumer_offsets-4 -> Vector(20), __consumer_offsets-7 -> Vector(20), __consumer_offsets-46 -> Vector(20), __consumer_offsets-25 -> Vector(20), __consumer_offsets-49 -> Vector(20), __consumer_offsets-16 -> Vector(20), test-0 -> Vector(20, 30, 10), __consumer_offsets-28 -> Vector(20), __consumer_offsets-31 -> Vector(20), test-2-2 -> Vector(20, 30), __consumer_offsets-37 -> Vector(20), filebeat-system-0 -> Vector(20), test-3-3 -> Vector(20, 30), __consumer_offsets-19 -> Vector(20), __consumer_offsets-13 -> Vector(20), __consumer_offsets-43 -> Vector(20), __consumer_offsets-1 -> Vector(20), __consumer_offsets-34 -> Vector(20), __consumer_offsets-10 -> Vector(20), test-3 -> Vector(20, 10, 30), __consumer_offsets-40 -> Vector(20)), 10 -> Map(__consumer_offsets-30 -> Vector(10), __consumer_offsets-21 -> Vector(10), __consumer_offsets-27 -> Vector(10), __consumer_offsets-9 -> Vector(10), __consumer_offsets-33 -> Vector(10), __consumer_offsets-36 -> Vector(10), __consumer_offsets-42 -> Vector(10), __consumer_offsets-3 -> Vector(10), __consumer_offsets-18 -> Vector(10), test-5 -> Vector(10, 30, 20), __consumer_offsets-15 -> Vector(10), __consumer_offsets-24 -> Vector(10), test-3-1 -> Vector(10, 20), __consumer_offsets-48 -> Vector(10), filebeat-kafka-0 -> Vector(10), __consumer_offsets-6 -> Vector(10), test-2-1 -> Vector(10, 20), test-3-2 -> Vector(10, 30), __consumer_offsets-0 -> Vector(10), __consumer_offsets-39 -> Vector(10), __consumer_offsets-12 -> Vector(10), __consumer_offsets-45 -> Vector(10), test-2 -> Vector(10, 20, 30)), 30 -> Map(__consumer_offsets-8 -> Vector(30), __consumer_offsets-35 -> Vector(30), __consumer_offsets-41 -> Vector(30), __consumer_offsets-23 -> Vector(30), __consumer_offsets-47 -> Vector(30), metricbeat-0 -> Vector(30), test-3-0 -> Vector(30, 10), metricbeat-kafka-0 -> Vector(30), filebeat-0 -> Vector(30), test-2-0 -> Vector(30, 10), __consumer_offsets-38 -> Vector(30), __consumer_offsets-17 -> Vector(30), test-1 -> Vector(30, 10, 20), test-2-3 -> Vector(30, 20), __consumer_offsets-11 -> Vector(30), __consumer_offsets-2 -> Vector(30), __consumer_offsets-14 -> Vector(30), test-4 -> Vector(30, 20, 10), metricbeat-system-0 -> Vector(30), __consumer_offsets-20 -> Vector(30), __consumer_offsets-44 -> Vector(30), __consumer_offsets-5 -> Vector(30), __consumer_offsets-26 -> Vector(30), __consumer_offsets-29 -> Vector(30), __consumer_offsets-32 -> Vector(30))) (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] DEBUG [Controller id=10] Topics not in preferred replica for broker 20 Map() (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] TRACE [Controller id=10] Leader imbalance ratio for broker 20 is 0.0 (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] DEBUG [Controller id=10] Topics not in preferred replica for broker 10 Map(test-3-1 -> Vector(10, 20)) (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] TRACE [Controller id=10] Leader imbalance ratio for broker 10 is 0.043478260869565216 (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] DEBUG [Controller id=10] Topics not in preferred replica for broker 30 Map() (kafka.controller.KafkaController)
+[2018-10-31 15:08:32,475] TRACE [Controller id=10] Leader imbalance ratio for broker 30 is 0.0 (kafka.controller.KafkaController)
+[2018-10-31 15:09:30,306] INFO [Controller id=10] New topics: [Set(foo)], deleted topics: [Set()], new partition replica assignment [Map(foo-0 -> Vector(20))] (kafka.controller.KafkaController)
+[2018-10-31 15:09:30,307] INFO [Controller id=10] New partition creation callback for foo-0 (kafka.controller.KafkaController)
+[2018-10-31 15:09:30,396] INFO [RequestSendThread controllerId=10] Controller 10 connected to 10.122.220.20:9094 (id: 20 rack: null) for sending state change requests (kafka.controller.RequestSendThread)
+[2018-10-31 15:09:30,397] INFO [RequestSendThread controllerId=10] Controller 10 connected to 10.122.220.20:9093 (id: 10 rack: null) for sending state change requests (kafka.controller.RequestSendThread)
+[2018-10-31 15:09:30,396] INFO [RequestSendThread controllerId=10] Controller 10 connected to 10.122.220.20:9095 (id: 30 rack: null) for sending state change requests (kafka.controller.RequestSendThread)
+[2018-10-31 15:13:32,475] TRACE [Controller id=10] Checking need to trigger auto leader balancing (kafka.controller.KafkaController)
+
diff --git a/filebeat/module/kafka/log/test/server-2.0.0.log b/filebeat/module/kafka/log/test/server-2.0.0.log
new file mode 100644
index 00000000000..425b68c79df
--- /dev/null
+++ b/filebeat/module/kafka/log/test/server-2.0.0.log
@@ -0,0 +1,160 @@
+[2018-10-17 12:04:41,718] INFO [GroupMetadataManager brokerId=10] Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
+[2018-10-17 12:14:41,719] INFO [GroupMetadataManager brokerId=10] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
+[2018-10-17 12:24:41,719] INFO [GroupMetadataManager brokerId=10] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
+[2018-10-17 12:34:41,719] INFO [GroupMetadataManager brokerId=10] Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
+[2018-10-17 12:44:41,719] INFO [GroupMetadataManager brokerId=10] Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
+[2018-10-17 12:50:23,313] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:50:23,314] INFO [Partition test-3 broker=10] test-3 starts at Leader Epoch 1 from offset 0. Previous Leader Epoch was: 0 (kafka.cluster.Partition)
+[2018-10-17 12:50:23,321] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:50:23,322] INFO [ReplicaFetcherManager on broker 10] Added fetcher for partitions List([test-0, initOffset 0 to broker BrokerEndPoint(30,10.122.220.20,9095)] ) (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:50:23,322] INFO [ReplicaAlterLogDirsManager on broker 10] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:50:23,323] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Shutting down (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:50:23,323] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Error sending fetch request (sessionId=1901923426, epoch=30531) to node 20: java.nio.channels.ClosedSelectorException. (org.apache.kafka.clients.FetchSessionHandler)
+[2018-10-17 12:50:23,324] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Stopped (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:50:23,331] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Shutdown completed (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:50:23,348] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-5,test-2,test-2-1 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:50:23,348] INFO [Partition test-5 broker=10] test-5 starts at Leader Epoch 1 from offset 0. Previous Leader Epoch was: 0 (kafka.cluster.Partition)
+[2018-10-17 12:50:23,350] INFO [Partition test-2 broker=10] test-2 starts at Leader Epoch 1 from offset 0. Previous Leader Epoch was: 0 (kafka.cluster.Partition)
+[2018-10-17 12:50:23,351] INFO [Partition test-2-1 broker=10] test-2-1 starts at Leader Epoch 1 from offset 0. Previous Leader Epoch was: 0 (kafka.cluster.Partition)
+[2018-10-17 12:50:23,355] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions  (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:50:23,360] INFO [ReplicaFetcherManager on broker 10] Added fetcher for partitions List() (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:50:23,361] INFO [ReplicaAlterLogDirsManager on broker 10] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:50:23,421] WARN [ReplicaFetcher replicaId=10, leaderId=30, fetcherId=0] Based on follower's leader epoch, leader replied with an unknown offset in test-0. The initial fetch offset 0 will be used for truncation. (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:50:23,421] INFO [Log partition=test-0, dir=/tmp/kafka-logs-10] Truncating to 0 has no effect as the largest offset in the log is -1 (kafka.log.Log)
+[2018-10-17 12:50:24,508] INFO [ReplicaAlterLogDirsManager on broker 10] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:51:56,064] INFO [Partition test-5 broker=10] Expanding ISR from 10,30 to 10,30,20 (kafka.cluster.Partition)
+[2018-10-17 12:51:56,091] INFO [Partition test-3 broker=10] Expanding ISR from 10,30 to 10,30,20 (kafka.cluster.Partition)
+[2018-10-17 12:51:56,098] INFO [Partition test-2 broker=10] Expanding ISR from 10,30 to 10,30,20 (kafka.cluster.Partition)
+[2018-10-17 12:51:56,104] INFO [Partition test-2-1 broker=10] Expanding ISR from 10 to 10,20 (kafka.cluster.Partition)
+[2018-10-17 12:54:31,461] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:54:31,481] INFO [ReplicaFetcherManager on broker 10] Added fetcher for partitions List([test-0, initOffset 0 to broker BrokerEndPoint(20,10.122.220.20,9094)] ) (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:54:31,482] INFO [ReplicaAlterLogDirsManager on broker 10] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:54:31,483] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Starting (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:54:31,501] WARN [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Based on follower's leader epoch, leader replied with an unknown offset in test-0. The initial fetch offset 0 will be used for truncation. (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:54:31,504] INFO [Log partition=test-0, dir=/tmp/kafka-logs-10] Truncating to 0 has no effect as the largest offset in the log is -1 (kafka.log.Log)
+[2018-10-17 12:54:31,504] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:54:31,508] INFO [ReplicaFetcherManager on broker 10] Added fetcher for partitions List([test-3, initOffset 0 to broker BrokerEndPoint(20,10.122.220.20,9094)] ) (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:54:31,510] INFO [ReplicaAlterLogDirsManager on broker 10] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:54:32,043] WARN [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Based on follower's leader epoch, leader replied with an unknown offset in test-3. The initial fetch offset 0 will be used for truncation. (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:54:32,044] INFO [Log partition=test-3, dir=/tmp/kafka-logs-10] Truncating to 0 has no effect as the largest offset in the log is -1 (kafka.log.Log)
+[2018-10-17 12:54:41,719] INFO [GroupMetadataManager brokerId=10] Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
+[2018-10-17 12:57:17,790] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3-2 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:57:17,809] INFO [Log partition=test-3-2, dir=/tmp/kafka-logs-10] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
+[2018-10-17 12:57:17,810] INFO [Log partition=test-3-2, dir=/tmp/kafka-logs-10] Completed load of log with 1 segments, log start offset 0 and log end offset 0 in 2 ms (kafka.log.Log)
+[2018-10-17 12:57:17,812] INFO Created log for partition test-3-2 in /tmp/kafka-logs-10 with properties {compression.type -> producer, message.format.version -> 2.0-IV1, file.delete.delay.ms -> 60000, max.message.bytes -> 1000012, min.compaction.lag.ms -> 0, message.timestamp.type -> CreateTime, message.downconversion.enable -> true, min.insync.replicas -> 1, segment.jitter.ms -> 0, preallocate -> false, min.cleanable.dirty.ratio -> 0.5, index.interval.bytes -> 4096, unclean.leader.election.enable -> false, retention.bytes -> -1, delete.retention.ms -> 86400000, cleanup.policy -> [delete], flush.ms -> 9223372036854775807, segment.ms -> 604800000, segment.bytes -> 1073741824, retention.ms -> 604800000, message.timestamp.difference.max.ms -> 9223372036854775807, segment.index.bytes -> 10485760, flush.messages -> 9223372036854775807}. (kafka.log.LogManager)
+[2018-10-17 12:57:17,816] INFO [Partition test-3-2 broker=10] No checkpointed highwatermark is found for partition test-3-2 (kafka.cluster.Partition)
+[2018-10-17 12:57:17,816] INFO Replica loaded for partition test-3-2 with initial high watermark 0 (kafka.cluster.Replica)
+[2018-10-17 12:57:17,816] INFO Replica loaded for partition test-3-2 with initial high watermark 0 (kafka.cluster.Replica)
+[2018-10-17 12:57:17,816] INFO [Partition test-3-2 broker=10] test-3-2 starts at Leader Epoch 0 from offset 0. Previous Leader Epoch was: -1 (kafka.cluster.Partition)
+[2018-10-17 12:57:17,817] INFO Replica loaded for partition test-3-0 with initial high watermark 0 (kafka.cluster.Replica)
+[2018-10-17 12:57:17,833] INFO [Log partition=test-3-0, dir=/tmp/kafka-logs-10] Loading producer state till offset 0 with message format version 2 (kafka.log.Log)
+[2018-10-17 12:57:17,833] INFO [Log partition=test-3-0, dir=/tmp/kafka-logs-10] Completed load of log with 1 segments, log start offset 0 and log end offset 0 in 5 ms (kafka.log.Log)
+[2018-10-17 12:57:17,835] INFO Created log for partition test-3-0 in /tmp/kafka-logs-10 with properties {compression.type -> producer, message.format.version -> 2.0-IV1, file.delete.delay.ms -> 60000, max.message.bytes -> 1000012, min.compaction.lag.ms -> 0, message.timestamp.type -> CreateTime, message.downconversion.enable -> true, min.insync.replicas -> 1, segment.jitter.ms -> 0, preallocate -> false, min.cleanable.dirty.ratio -> 0.5, index.interval.bytes -> 4096, unclean.leader.election.enable -> false, retention.bytes -> -1, delete.retention.ms -> 86400000, cleanup.policy -> [delete], flush.ms -> 9223372036854775807, segment.ms -> 604800000, segment.bytes -> 1073741824, retention.ms -> 604800000, message.timestamp.difference.max.ms -> 9223372036854775807, segment.index.bytes -> 10485760, flush.messages -> 9223372036854775807}. (kafka.log.LogManager)
+[2018-10-17 12:57:17,836] INFO [Partition test-3-0 broker=10] No checkpointed highwatermark is found for partition test-3-0 (kafka.cluster.Partition)
+[2018-10-17 12:57:17,836] INFO Replica loaded for partition test-3-0 with initial high watermark 0 (kafka.cluster.Replica)
+[2018-10-17 12:57:17,837] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:57:17,838] INFO [ReplicaFetcherManager on broker 10] Added fetcher for partitions List([test-3-0, initOffset 0 to broker BrokerEndPoint(30,10.122.220.20,9095)] ) (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:57:17,839] INFO [ReplicaAlterLogDirsManager on broker 10] Added fetcher for partitions List() (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:57:17,896] WARN [ReplicaFetcher replicaId=10, leaderId=30, fetcherId=0] Based on follower's leader epoch, leader replied with an unknown offset in test-3-0. The initial fetch offset 0 will be used for truncation. (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:57:17,897] INFO [Log partition=test-3-0, dir=/tmp/kafka-logs-10] Truncating to 0 has no effect as the largest offset in the log is -1 (kafka.log.Log)
+[2018-10-17 12:57:18,400] ERROR [ReplicaFetcher replicaId=10, leaderId=30, fetcherId=0] Error for partition test-3-0 at offset 0 (kafka.server.ReplicaFetcherThread)
+org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
+[2018-10-17 12:58:47,490] INFO Terminating process due to signal SIGTERM (org.apache.kafka.common.utils.LoggingSignalHandler)
+[2018-10-17 12:58:47,492] INFO [KafkaServer id=10] shutting down (kafka.server.KafkaServer)
+[2018-10-17 12:58:47,494] INFO [KafkaServer id=10] Starting controlled shutdown (kafka.server.KafkaServer)
+[2018-10-17 12:58:47,547] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions  (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,550] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions  (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,556] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,556] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-0 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,558] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,558] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-3-0 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,561] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-2-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,561] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-2-0 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,567] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-1 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,567] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-1 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,568] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-4 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,568] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-4 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,568] INFO [ReplicaFetcher replicaId=10, leaderId=30, fetcherId=0] Shutting down (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:58:47,577] INFO [ReplicaFetcher replicaId=10, leaderId=30, fetcherId=0] Error sending fetch request (sessionId=461323381, epoch=31537) to node 30: java.nio.channels.ClosedSelectorException. (org.apache.kafka.clients.FetchSessionHandler)
+[2018-10-17 12:58:47,577] INFO [ReplicaFetcher replicaId=10, leaderId=30, fetcherId=0] Stopped (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:58:47,583] INFO [ReplicaFetcher replicaId=10, leaderId=30, fetcherId=0] Shutdown completed (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:58:47,585] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,586] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-3 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,594] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Shutting down (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:58:47,601] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Error sending fetch request (sessionId=1016438239, epoch=510) to node 20: java.nio.channels.ClosedSelectorException. (org.apache.kafka.clients.FetchSessionHandler)
+[2018-10-17 12:58:47,602] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Stopped (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:58:47,602] INFO [ReplicaFetcher replicaId=10, leaderId=20, fetcherId=0] Shutdown completed (kafka.server.ReplicaFetcherThread)
+[2018-10-17 12:58:47,604] INFO [KafkaServer id=10] Controlled shutdown succeeded (kafka.server.KafkaServer)
+[2018-10-17 12:58:47,605] INFO [/config/changes-event-process-thread]: Shutting down (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
+[2018-10-17 12:58:47,606] INFO [/config/changes-event-process-thread]: Stopped (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
+[2018-10-17 12:58:47,606] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions  (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,606] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions  (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,606] INFO [/config/changes-event-process-thread]: Shutdown completed (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
+[2018-10-17 12:58:47,607] INFO [SocketServer brokerId=10] Stopping socket server request processors (kafka.network.SocketServer)
+[2018-10-17 12:58:47,608] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,608] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-3 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,609] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-1 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,609] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-1 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,610] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,610] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-0 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,611] INFO [ReplicaFetcherManager on broker 10] Removed fetcher for partitions test-3-0 (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,611] INFO [ReplicaAlterLogDirsManager on broker 10] Removed fetcher for partitions test-3-0 (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,617] INFO [SocketServer brokerId=10] Stopped socket server request processors (kafka.network.SocketServer)
+[2018-10-17 12:58:47,631] INFO [Kafka Request Handler on Broker 10], shutting down (kafka.server.KafkaRequestHandlerPool)
+[2018-10-17 12:58:47,637] INFO [Kafka Request Handler on Broker 10], shut down completely (kafka.server.KafkaRequestHandlerPool)
+[2018-10-17 12:58:47,647] INFO [KafkaApi-10] Shutdown complete. (kafka.server.KafkaApis)
+[2018-10-17 12:58:47,649] INFO [ExpirationReaper-10-topic]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,846] INFO [ExpirationReaper-10-topic]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,846] INFO [ExpirationReaper-10-topic]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,847] INFO [TransactionCoordinator id=10] Shutting down. (kafka.coordinator.transaction.TransactionCoordinator)
+[2018-10-17 12:58:47,847] INFO [ProducerId Manager 10]: Shutdown complete: last producerId assigned 2000 (kafka.coordinator.transaction.ProducerIdManager)
+[2018-10-17 12:58:47,848] INFO [Transaction State Manager 10]: Shutdown complete (kafka.coordinator.transaction.TransactionStateManager)
+[2018-10-17 12:58:47,848] INFO [Transaction Marker Channel Manager 10]: Shutting down (kafka.coordinator.transaction.TransactionMarkerChannelManager)
+[2018-10-17 12:58:47,848] INFO [Transaction Marker Channel Manager 10]: Stopped (kafka.coordinator.transaction.TransactionMarkerChannelManager)
+[2018-10-17 12:58:47,848] INFO [Transaction Marker Channel Manager 10]: Shutdown completed (kafka.coordinator.transaction.TransactionMarkerChannelManager)
+[2018-10-17 12:58:47,848] INFO [TransactionCoordinator id=10] Shutdown complete. (kafka.coordinator.transaction.TransactionCoordinator)
+[2018-10-17 12:58:47,849] INFO [GroupCoordinator 10]: Shutting down. (kafka.coordinator.group.GroupCoordinator)
+[2018-10-17 12:58:47,849] INFO [ExpirationReaper-10-Heartbeat]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,959] INFO [ExpirationReaper-10-Heartbeat]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,959] INFO [ExpirationReaper-10-Heartbeat]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,960] INFO [ExpirationReaper-10-Rebalance]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,964] INFO [ExpirationReaper-10-Rebalance]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,964] INFO [ExpirationReaper-10-Rebalance]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:47,964] INFO [GroupCoordinator 10]: Shutdown complete. (kafka.coordinator.group.GroupCoordinator)
+[2018-10-17 12:58:47,965] INFO [ReplicaManager broker=10] Shutting down (kafka.server.ReplicaManager)
+[2018-10-17 12:58:47,965] INFO [LogDirFailureHandler]: Shutting down (kafka.server.ReplicaManager$LogDirFailureHandler)
+[2018-10-17 12:58:47,968] INFO [LogDirFailureHandler]: Stopped (kafka.server.ReplicaManager$LogDirFailureHandler)
+[2018-10-17 12:58:47,968] INFO [LogDirFailureHandler]: Shutdown completed (kafka.server.ReplicaManager$LogDirFailureHandler)
+[2018-10-17 12:58:47,970] INFO [ReplicaFetcherManager on broker 10] shutting down (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,972] INFO [ReplicaFetcherManager on broker 10] shutdown completed (kafka.server.ReplicaFetcherManager)
+[2018-10-17 12:58:47,973] INFO [ReplicaAlterLogDirsManager on broker 10] shutting down (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,974] INFO [ReplicaAlterLogDirsManager on broker 10] shutdown completed (kafka.server.ReplicaAlterLogDirsManager)
+[2018-10-17 12:58:47,974] INFO [ExpirationReaper-10-Fetch]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,059] INFO [ExpirationReaper-10-Fetch]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,059] INFO [ExpirationReaper-10-Fetch]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,060] INFO [ExpirationReaper-10-Produce]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,184] INFO [ExpirationReaper-10-Produce]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,184] INFO [ExpirationReaper-10-Produce]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,184] INFO [ExpirationReaper-10-DeleteRecords]: Shutting down (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,184] INFO [ExpirationReaper-10-DeleteRecords]: Shutdown completed (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,184] INFO [ExpirationReaper-10-DeleteRecords]: Stopped (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
+[2018-10-17 12:58:48,186] INFO [ReplicaManager broker=10] Shut down completely (kafka.server.ReplicaManager)
+[2018-10-17 12:58:48,187] INFO Shutting down. (kafka.log.LogManager)
+[2018-10-17 12:58:48,208] INFO [ProducerStateManager partition=filebeat-kafka-0] Writing producer snapshot at offset 771608 (kafka.log.ProducerStateManager)
+[2018-10-17 12:58:48,244] INFO Shutdown complete. (kafka.log.LogManager)
+[2018-10-17 12:58:48,248] INFO [ZooKeeperClient] Closing. (kafka.zookeeper.ZooKeeperClient)
+[2018-10-17 12:58:48,251] INFO Session: 0x100171af0080002 closed (org.apache.zookeeper.ZooKeeper)
+[2018-10-17 12:58:48,252] INFO [ZooKeeperClient] Closed. (kafka.zookeeper.ZooKeeperClient)
+[2018-10-17 12:58:48,253] INFO [ThrottledChannelReaper-Fetch]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:48,253] INFO EventThread shut down for session: 0x100171af0080002 (org.apache.zookeeper.ClientCnxn)
+[2018-10-17 12:58:49,201] INFO [ThrottledChannelReaper-Fetch]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:49,201] INFO [ThrottledChannelReaper-Fetch]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:49,201] INFO [ThrottledChannelReaper-Produce]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:49,209] INFO [ThrottledChannelReaper-Produce]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:49,209] INFO [ThrottledChannelReaper-Produce]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:49,209] INFO [ThrottledChannelReaper-Request]: Shutting down (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:50,209] INFO [ThrottledChannelReaper-Request]: Stopped (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:50,209] INFO [ThrottledChannelReaper-Request]: Shutdown completed (kafka.server.ClientQuotaManager$ThrottledChannelReaper)
+[2018-10-17 12:58:50,209] INFO [SocketServer brokerId=10] Shutting down socket server (kafka.network.SocketServer)
+[2018-10-17 12:58:50,224] INFO [SocketServer brokerId=10] Shutdown completed (kafka.network.SocketServer)
+[2018-10-17 12:58:50,233] INFO [KafkaServer id=10] shut down completed (kafka.server.KafkaServer)
diff --git a/filebeat/module/kafka/log/test/state-change-2.0.0.log b/filebeat/module/kafka/log/test/state-change-2.0.0.log
new file mode 100644
index 00000000000..888c381706b
--- /dev/null
+++ b/filebeat/module/kafka/log/test/state-change-2.0.0.log
@@ -0,0 +1,2 @@
+[2018-10-31 15:09:30,451] TRACE [Broker id=20] Cached leader info PartitionState(controllerEpoch=5, leader=20, leaderEpoch=0, isr=[20], zkVersion=0, replicas=[20], offlineReplicas=[]) for partition foo-0 in response to UpdateMetadata request sent by controller 10 epoch 5 with correlation id 146 (state.change.logger)
+
diff --git a/filebeat/module/kafka/log/test/state-change-2.0.0.log-expected.json b/filebeat/module/kafka/log/test/state-change-2.0.0.log-expected.json
new file mode 100644
index 00000000000..3cfb112c20e
--- /dev/null
+++ b/filebeat/module/kafka/log/test/state-change-2.0.0.log-expected.json
@@ -0,0 +1,18 @@
+[
+    {
+        "@timestamp": "2018-10-31T15:09:30.451Z", 
+        "fileset.module": "kafka", 
+        "fileset.name": "log", 
+        "input.type": "log", 
+        "kafka.log.class": "state.change.logger", 
+        "kafka.log.component": "Broker id=20", 
+        "kafka.log.level": "TRACE", 
+        "kafka.log.message": "Cached leader info PartitionState(controllerEpoch=5, leader=20, leaderEpoch=0, isr=[20], zkVersion=0, replicas=[20], offlineReplicas=[]) for partition foo-0 in response to UpdateMetadata request sent by controller 10 epoch 5 with correlation id 146", 
+        "log.flags": [
+            "multiline"
+        ], 
+        "message": "[2018-10-31 15:09:30,451] TRACE [Broker id=20] Cached leader info PartitionState(controllerEpoch=5, leader=20, leaderEpoch=0, isr=[20], zkVersion=0, replicas=[20], offlineReplicas=[]) for partition foo-0 in response to UpdateMetadata request sent by controller 10 epoch 5 with correlation id 146 (state.change.logger)\n", 
+        "offset": 0, 
+        "prospector.type": "log"
+    }
+]
\ No newline at end of file