diff --git a/collector/libbeat.go b/collector/libbeat.go
index 292ad30..c6c527c 100644
--- a/collector/libbeat.go
+++ b/collector/libbeat.go
@@ -15,6 +15,7 @@ type LibBeat struct {
 		Reloads float64 `json:"reloads"`
 	} `json:"config"`
 	Output   LibBeatOutput   `json:"output"`
+	Outputs  LibBeatOutputs  `json:"outputs"`
 	Pipeline LibBeatPipeline `json:"pipeline"`
 }
 
@@ -45,6 +46,14 @@ type LibBeatOutput struct {
 	Type   string                   `json:"type"`
 }
 
+//LibBeatOutputs json structure
+type LibBeatOutputs struct {
+	Kafka struct {
+		Read  float64 `json:"bytes_read"`
+		Write float64 `json:"bytes_write"`
+	} `json:"kafka"`
+}
+
 //LibBeatPipeline json structure
 type LibBeatPipeline struct {
 	Clients float64       `json:"clients"`
@@ -119,7 +128,12 @@ func NewLibBeatCollector(beatInfo *BeatInfo, stats *Stats) prometheus.Collector
 					nil, nil,
 				),
 				eval: func(stats *Stats) float64 {
-					return stats.LibBeat.Output.Read.Bytes
+					if stats.LibBeat.Output.Type == "kafka" {
+						return stats.LibBeat.Outputs.Kafka.Read
+					} else {
+						return stats.LibBeat.Output.Read.Bytes
+					}
+
 				},
 				valType: prometheus.CounterValue,
 			},
@@ -141,7 +155,11 @@ func NewLibBeatCollector(beatInfo *BeatInfo, stats *Stats) prometheus.Collector
 					nil, nil,
 				),
 				eval: func(stats *Stats) float64 {
-					return stats.LibBeat.Output.Write.Bytes
+					if stats.LibBeat.Output.Type == "kafka" {
+						return stats.LibBeat.Outputs.Kafka.Write
+					} else {
+						return stats.LibBeat.Output.Write.Bytes
+					}
 				},
 				valType: prometheus.CounterValue,
 			},