Add owner tag on partitions in burrow input (#4281)

plugins/inputs/burrow/README.md

@@ -92,6 +92,7 @@ Supported Burrow version: `1.x`
   - group (string)
   - topic (string)
   - partition (int)
+  - owner (string)
 
 * `burrow_topic`
   - cluster (string)

plugins/inputs/burrow/burrow.go

@@ -116,6 +116,7 @@ type (
 		Start      apiStatusResponseLagItem `json:"start"`
 		End        apiStatusResponseLagItem `json:"end"`
 		CurrentLag int64                    `json:"current_lag"`
+		Owner      string                   `json:"owner"`
 	}
 
 	// response: lag field item
@@ -447,6 +448,7 @@ func (b *burrow) genGroupLagMetrics(r *apiResponse, cluster, group string, acc t
 				"group":     group,
 				"topic":     partition.Topic,
 				"partition": strconv.FormatInt(int64(partition.Partition), 10),
+				"owner":     partition.Owner,
 			},
 		)
 	}

plugins/inputs/burrow/burrow_test.go

@@ -129,9 +129,9 @@ func TestBurrowPartition(t *testing.T) {
 		},
 	}
 	tags := []map[string]string{
-		{"cluster": "clustername1", "group": "group1", "topic": "topicA", "partition": "0"},
+		{"cluster": "clustername1", "group": "group1", "topic": "topicA", "partition": "0", "owner": "kafka1"},
-		{"cluster": "clustername1", "group": "group1", "topic": "topicA", "partition": "1"},
+		{"cluster": "clustername1", "group": "group1", "topic": "topicA", "partition": "1", "owner": "kafka2"},
-		{"cluster": "clustername1", "group": "group1", "topic": "topicA", "partition": "2"},
+		{"cluster": "clustername1", "group": "group1", "topic": "topicA", "partition": "2", "owner": "kafka3"},
 	}
 
 	require.Empty(t, acc.Errors)

plugins/inputs/burrow/testdata/v3_kafka_clustername1_consumer_group1_lag.json

@@ -10,7 +10,7 @@
       {
         "topic": "topicA",
         "partition": 0,
-        "owner": "kafka",
+        "owner": "kafka1",
         "status": "OK",
         "start": {
           "offset": 431323195,
@@ -28,7 +28,7 @@
       {
         "topic": "topicA",
         "partition": 1,
-        "owner": "kafka",
+        "owner": "kafka2",
         "status": "OK",
         "start": {
           "offset": 431322962,
@@ -46,7 +46,7 @@
       {
         "topic": "topicA",
         "partition": 2,
-        "owner": "kafka",
+        "owner": "kafka3",
         "status": "OK",
         "start": {
           "offset": 428636563,