fix: remove message present in the ivalid partitions

Author: siddhanta-rath

publisher/kafka.go

@@ -158,7 +158,10 @@ func (pr *Kafka) reportBatchMetrics(stats map[string]interface{}) {
 		// Sum messages across all partitions
 		msgCnt := 0.0
 		if partitions, ok := topicStats["partitions"].(map[string]interface{}); ok {
-			for _, p := range partitions {
+			for partition, p := range partitions {
+				if partition == "-1" { // ignore invalid partition
+					continue
+				}
 				partStats, ok := p.(map[string]interface{})
 				if !ok {
 					continue
@@ -185,8 +188,8 @@ func (pr *Kafka) reportBatchMetrics(stats map[string]interface{}) {
 		}
 
 		// Emit metrics
-		metrics.Count("kafka_producer_batch_count_total", int(batchCnt), fmt.Sprintf("topic=%s", topicName))
-		metrics.Count("kafka_producer_message_count_total", int(msgCnt), fmt.Sprintf("topic=%s", topicName))
+		metrics.Gauge("kafka_producer_batch_count_total", int(batchCnt), fmt.Sprintf("topic=%s", topicName))
+		metrics.Gauge("kafka_producer_message_count_total", int(msgCnt), fmt.Sprintf("topic=%s", topicName))
 		metrics.Gauge("kafka_producer_batch_size_avg_bytes", batchSizeAvg, fmt.Sprintf("topic=%s", topicName))
 		metrics.Gauge("kafka_producer_messages_per_batch_avg", msgsPerBatch, fmt.Sprintf("topic=%s", topicName))