Fix publish_outcomes and publish_errors_total metrics

internal/consumer/consumer.go

@@ -271,7 +271,6 @@ func (c *KafkaConsumer) pollLoop() {
 					"offset":    record.Offset,
 				})
 			}
-			c.handleOutcome(outcome, err, record)
 
 			c.metricEmitter.Notify(metrics.Event{
 				Name:  metrics.PublisherOutcomes,
@@ -280,6 +279,8 @@ func (c *KafkaConsumer) pollLoop() {
 					metrics.OutcomeLabel, outcome.String(),
 				},
 			})
+
+			c.handleOutcome(outcome, err, record)
 		})
 	}
 }

internal/consumer/handler.go

@@ -161,7 +161,7 @@ func (h *WRPMessageHandler) HandleMessage(ctx context.Context, record *kgo.Recor
 	}
 
 	h.emitLog(log.LevelDebug, "successfully routed WRP message", map[string]any{
-		"outcome": outcome,
+		"outcome": outcome.String(),
 	})
 
 	return getOutcome(outcome), nil

internal/publisher/publisher.go

@@ -202,7 +202,7 @@ func (p *KafkaPublisher) Produce(ctx context.Context, msg *wrp.Message) (wrpkafk
 				metrics.ErrorTypeLabel, "not_started",
 			},
 		})
-		return 0, ErrPublisherNotStarted // Return 0 (which corresponds to wrpkafka.Accepted) as default
+		return 0, ErrPublisherNotStarted // Return 0 (which corresponds to wrpkafka.Attempted) as default
 	}
 
 	outcome, err := p.wrpPublisher.Produce(ctx, msg)
@@ -213,6 +213,13 @@ func (p *KafkaPublisher) Produce(ctx context.Context, msg *wrp.Message) (wrpkafk
 			"source":       msg.Source,
 			"destination":  msg.Destination,
 		}))
+		p.metricEmitter.Notify(metrics.Event{
+			Name:  metrics.PublisherErrorsCounter,
+			Value: 1,
+			Labels: []string{
+				metrics.ErrorTypeLabel, "failed_to_produce_message",
+			},
+		})
 		return outcome, fmt.Errorf("failed to produce message: %w", err)
 	}