Make error fields optional in partition event (elastic#3089) (elastic…

…#3092) * Update data.json (cherry picked from commit 298416e)
leweafan · Dec 1, 2016 · 391f3e7 · 391f3e7
1 parent 94a6f83
commit 391f3e7
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 15 deletions.
diff --git a/metricbeat/module/kafka/partition/_meta/data.json b/metricbeat/module/kafka/partition/_meta/data.json
@@ -15,14 +15,13 @@
                 "oldest": 0
             },
             "partition": {
-                "error": 0,
                 "id": 0,
                 "insync_replica": true,
                 "leader": 0,
                 "replica": 0
             },
             "topic": {
-                "name": "test-metricbeat-8760238589576171408"
+                "name": "test-metricbeat-70692474374989458"
             }
         }
     },

diff --git a/metricbeat/module/kafka/partition/partition.go b/metricbeat/module/kafka/partition/partition.go
@@ -130,9 +130,12 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 	for _, topic := range response.Topics {
 		evtTopic := common.MapStr{
 			"name": topic.Name,
-			"error": common.MapStr{
+		}
+
+		if topic.Err != 0 {
+			evtTopic["error"] = common.MapStr{
 				"code": topic.Err,
-			},
+			}
 		}
 
 		for _, partition := range topic.Partitions {
@@ -157,19 +160,24 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 					continue
 				}
 
+				partitionEvent := common.MapStr{
+					"id":             partition.ID,
+					"leader":         partition.Leader,
+					"replica":        id,
+					"insync_replica": hasID(id, partition.Isr),
+				}
+
+				if partition.Err != 0 {
+					partitionEvent["error"] = common.MapStr{
+						"code": partition.Err,
+					}
+				}
+
 				// create event
 				event := common.MapStr{
-					"topic":  evtTopic,
-					"broker": evtBroker,
-					"partition": common.MapStr{
-						"id": partition.ID,
-						"error": common.MapStr{
-							"code": partition.Err,
-						},
-						"leader":         partition.Leader,
-						"replica":        id,
-						"insync_replica": hasID(id, partition.Isr),
-					},
+					"topic":     evtTopic,
+					"broker":    evtBroker,
+					"partition": partitionEvent,
 					"offset": common.MapStr{
 						"newest": offNewest,
 						"oldest": offOldest,