Kafka producer metric topic label fix (#432)

beatlabs · Oct 29, 2021 · 7b51009 · 7b51009
1 parent 0f4ea6a
commit 7b51009
Showing 1 changed file with 9 additions and 3 deletions.
diff --git a/client/kafka/v2/sync_producer.go b/client/kafka/v2/sync_producer.go
@@ -57,19 +57,19 @@ func (p *SyncProducer) SendBatch(ctx context.Context, messages []*sarama.Produce
 
 	for _, msg := range messages {
 		if err := injectTracingHeaders(msg, sp); err != nil {
-			statusCountAdd(deliveryTypeSync, deliveryStatusSendError, batchTarget, len(messages))
+			statusCountAdd(deliveryTypeSync, deliveryStatusSendError, msg.Topic, len(messages))
 			trace.SpanError(sp)
 			return fmt.Errorf("failed to inject tracing headers: %w", err)
 		}
 	}
 
 	if err := p.syncProd.SendMessages(messages); err != nil {
-		statusCountAdd(deliveryTypeSync, deliveryStatusSendError, batchTarget, len(messages))
+		statusCountBatchAdd(deliveryTypeSync, deliveryStatusSendError, messages)
 		trace.SpanError(sp)
 		return err
 	}
 
-	statusCountAdd(deliveryTypeSync, deliveryStatusSent, batchTarget, len(messages))
+	statusCountBatchAdd(deliveryTypeSync, deliveryStatusSent, messages)
 	trace.SpanSuccess(sp)
 	return nil
 }
@@ -86,3 +86,9 @@ func (p *SyncProducer) Close() error {
 	}
 	return nil
 }
+
+func statusCountBatchAdd(deliveryType string, status deliveryStatus, messages []*sarama.ProducerMessage) {
+	for _, msg := range messages {
+		statusCountAdd(deliveryType, status, msg.Topic, 1)
+	}
+}