Fix recording issue

utr1903 · Feb 4, 2024 · 06d7359 · 06d7359
1 parent a67edcd
commit 06d7359
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/apps/golang/commons/otel/kafka/kafka_consumer_interceptor.go b/apps/golang/commons/otel/kafka/kafka_consumer_interceptor.go
@@ -105,6 +105,7 @@ type KafkaConsumeTelemetryContext struct {
 }
 
 func (k *KafkaConsumeTelemetryContext) End(
+	ctx context.Context,
 	err error,
 ) {
 	elapsedTime := float64(time.Since(k.startTime)) / float64(time.Millisecond)
@@ -113,7 +114,7 @@ func (k *KafkaConsumeTelemetryContext) End(
 	if err != nil {
 		attrs = append(attrs, semconv.ErrorType.String(err.Error()))
 	}
-	k.latency.Record(k.ctx, elapsedTime,
+	k.latency.Record(ctx, elapsedTime,
 		metric.WithAttributes(
 			attrs...,
 		))

diff --git a/apps/golang/kafkaconsumer/consumer/consumer.go b/apps/golang/kafkaconsumer/consumer/consumer.go
@@ -213,15 +213,15 @@ func (g *groupHandler) consumeMessage(
 	err := g.storeIntoDb(ctx, name)
 	if err != nil {
 		g.logger.Log(logrus.ErrorLevel, ctx, name, "Consuming message is failed.")
-		kctCtx.End(err)
+		kctCtx.End(ctx, err)
 		return nil
 	}
 
 	// Acknowledge message
 	session.MarkMessage(msg, "")
 	g.logger.Log(logrus.InfoLevel, ctx, name, "Consuming message is succeeded.")
 
-	kctCtx.End(nil)
+	kctCtx.End(ctx, nil)
 	return nil
 }