[receiver/kafkareceiver] Add 'topic' attribute to kafka internal metrics (#39550)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
Add 'topic' attribute to kafka existing internal metrics.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes #35336

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Updated the existing `kafkareceiver_test.go`

<!--Describe the documentation added.-->
#### Documentation
Changelog

<!--Please delete paragraphs that you did not use before submitting.-->

Author: petern48

.chloggen/kafka_topic_metric_attr.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Added a new 'topic' attribute to all existing internal consume-claim metrics.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35336]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/kafkareceiver/kafka_receiver.go

@@ -35,6 +35,7 @@ const (
 	transport = "kafka"
 	// TODO: update the following attributes to reflect semconv
 	attrInstanceName = "name"
+	attrTopic        = "topic"
 	attrPartition    = "partition"
 )
 
@@ -78,7 +79,10 @@ func newLogsReceiver(config *Config, set receiver.Settings, nextConsumer consume
 			config, set.Logger,
 			c.telemetryBuilder.KafkaReceiverUnmarshalFailedLogRecords,
 			[]metric.AddOption{
-				metric.WithAttributes(attribute.String(attrInstanceName, c.id.String())),
+				metric.WithAttributeSet(attribute.NewSet(
+					attribute.String(attrInstanceName, c.id.String()),
+					attribute.String(attrTopic, config.Logs.Topic),
+				)),
 			},
 			&logsHandler{
 				unmarshaler: unmarshaler,
@@ -101,7 +105,10 @@ func newMetricsReceiver(config *Config, set receiver.Settings, nextConsumer cons
 			config, set.Logger,
 			c.telemetryBuilder.KafkaReceiverUnmarshalFailedMetricPoints,
 			[]metric.AddOption{
-				metric.WithAttributes(attribute.String(attrInstanceName, c.id.String())),
+				metric.WithAttributeSet(attribute.NewSet(
+					attribute.String(attrInstanceName, c.id.String()),
+					attribute.String(attrTopic, config.Metrics.Topic),
+				)),
 			},
 			&metricsHandler{
 				unmarshaler: unmarshaler,
@@ -124,7 +131,10 @@ func newTracesReceiver(config *Config, set receiver.Settings, nextConsumer consu
 			config, set.Logger,
 			c.telemetryBuilder.KafkaReceiverUnmarshalFailedSpans,
 			[]metric.AddOption{
-				metric.WithAttributes(attribute.String(attrInstanceName, c.id.String())),
+				metric.WithAttributeSet(attribute.NewSet(
+					attribute.String(attrInstanceName, c.id.String()),
+					attribute.String(attrTopic, config.Traces.Topic),
+				)),
 			},
 			&tracesHandler{
 				unmarshaler: unmarshaler,
@@ -155,6 +165,7 @@ func newMessageHandlerConsumeFunc[T plog.Logs | pmetric.Metrics | ptrace.Traces]
 		data, n, err = h.unmarshalData(message.Value)
 		if err != nil {
 			logger.Error("failed to unmarshal message", zap.Error(err))
+			metricAddOpts = append(metricAddOpts, metric.WithAttributes(attribute.String(attrPartition, strconv.Itoa(int(message.Partition)))))
 			unmarshalFailedCounter.Add(ctx, 1, metricAddOpts...)
 			return err
 		}
@@ -487,6 +498,7 @@ func (c *consumerGroupHandler) handleMessage(
 	ctx := newContextWithHeaders(session.Context(), message.Headers)
 	attrs := attribute.NewSet(
 		attribute.String(attrInstanceName, c.id.String()),
+		attribute.String(attrTopic, message.Topic),
 		attribute.String(attrPartition, strconv.Itoa(int(claim.Partition()))),
 	)
 	c.telemetryBuilder.KafkaReceiverMessages.Add(ctx, 1, metric.WithAttributeSet(attrs))

receiver/kafkareceiver/kafka_receiver_test.go

@@ -285,6 +285,8 @@ func TestReceiver_InternalTelemetry(t *testing.T) {
 		Value: 1,
 		Attributes: attribute.NewSet(
 			attribute.String("name", set.ID.String()),
+			attribute.String("topic", "otlp_spans"),
+			attribute.String("partition", "0"),
 		),
 	}}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
 
@@ -301,6 +303,7 @@ func TestReceiver_InternalTelemetry(t *testing.T) {
 		Value: 5,
 		Attributes: attribute.NewSet(
 			attribute.String("name", set.ID.String()),
+			attribute.String("topic", "otlp_spans"),
 			attribute.String("partition", "0"),
 		),
 	}}, metricdatatest.IgnoreTimestamp())
@@ -327,6 +330,7 @@ func TestReceiver_InternalTelemetry(t *testing.T) {
 		Value: 4, // offset of the final message
 		Attributes: attribute.NewSet(
 			attribute.String("name", set.ID.String()),
+			attribute.String("topic", "otlp_spans"),
 			attribute.String("partition", "0"),
 		),
 	}}, metricdatatest.IgnoreTimestamp())
@@ -335,6 +339,7 @@ func TestReceiver_InternalTelemetry(t *testing.T) {
 		Value: 0,
 		Attributes: attribute.NewSet(
 			attribute.String("name", set.ID.String()),
+			attribute.String("topic", "otlp_spans"),
 			attribute.String("partition", "0"),
 		),
 	}}, metricdatatest.IgnoreTimestamp())
@@ -383,6 +388,8 @@ func TestNewLogsReceiver(t *testing.T) {
 		Value: 1,
 		Attributes: attribute.NewSet(
 			attribute.String("name", set.ID.String()),
+			attribute.String("topic", "otlp_logs"),
+			attribute.String("partition", "0"),
 		),
 	}}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
 
@@ -435,6 +442,8 @@ func TestNewMetricsReceiver(t *testing.T) {
 		Value: 1,
 		Attributes: attribute.NewSet(
 			attribute.String("name", set.ID.String()),
+			attribute.String("topic", "otlp_metrics"),
+			attribute.String("partition", "0"),
 		),
 	}}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())