kafkareceiver: propagate Kafka headers as metadata

Similar to open-telemetry#39132, but for the Kafka receiver, updates the receiver to
propagate Kafka headers as client.Info (metadata). Allowing downstream
processors and exporters to access the values via the enriched context.

Closes open-telemetry#39129

Signed-off-by: Marc Lopez Rubio <[email protected]>

Author: marclop

.chloggen/f_propagate-kafka-headers-as-metadata.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: Propagate Kafka headers as metadata
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39129]
+
+# (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: Allwos the Kafka receiver to propagate Kafka headers as client.Info (metadata). Allowing downstream processors and exporters to access the values via the enriched context.
+
+# 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/README.md

@@ -17,6 +17,8 @@ Kafka receiver receives traces, metrics, and logs from Kafka. Message payload en
 
 Note that metrics and logs only support OTLP.
 
+If used in conjunction with the `kafkaexporter` configured with `include_metadata_keys`. The Kafka receiver will also propagate the Kafka headers to the downstream pipeline, giving access to the rest of the pipeline to arbitrary metadata keys and values.
+
 ## Getting Started
 
 There are no required settings.

receiver/kafkareceiver/go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin v0.123.0
 	github.com/openzipkin/zipkin-go v0.4.3
 	github.com/stretchr/testify v1.10.0
+	go.opentelemetry.io/collector/client v1.29.1-0.20250402200755-cb5c3f4fb9dc
 	go.opentelemetry.io/collector/component v1.29.1-0.20250402200755-cb5c3f4fb9dc
 	go.opentelemetry.io/collector/component/componenttest v0.123.1-0.20250402200755-cb5c3f4fb9dc
 	go.opentelemetry.io/collector/config/configretry v1.29.1-0.20250402200755-cb5c3f4fb9dc

receiver/kafkareceiver/go.sum

@@ -13,6 +13,7 @@ import (
 
 	"github.com/IBM/sarama"
 	"github.com/cenkalti/backoff/v4"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/consumer"
@@ -539,7 +540,10 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 				session.MarkMessage(message, "")
 			}
 
-			ctx := c.obsrecv.StartTracesOp(session.Context())
+			// If the Kafka exporter has propagated headers in the message,
+			// create a new context with client.Info in it.
+			ctx := newContextWithHeaders(session.Context(), message.Headers)
+			ctx = c.obsrecv.StartTracesOp(ctx)
 			attrs := attribute.NewSet(
 				attribute.String(attrInstanceName, c.id.String()),
 				attribute.String(attrPartition, strconv.Itoa(int(claim.Partition()))),
@@ -560,7 +564,7 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 
 			c.headerExtractor.extractHeadersTraces(traces, message)
 			spanCount := traces.SpanCount()
-			err = c.nextConsumer.ConsumeTraces(session.Context(), traces)
+			err = c.nextConsumer.ConsumeTraces(ctx, traces)
 			c.obsrecv.EndTracesOp(ctx, c.unmarshaler.Encoding(), spanCount, err)
 			if err != nil {
 				if errorRequiresBackoff(err) && c.backOff != nil {
@@ -653,7 +657,10 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 				session.MarkMessage(message, "")
 			}
 
-			ctx := c.obsrecv.StartMetricsOp(session.Context())
+			// If the Kafka exporter has propagated headers in the message,
+			// create a new context with client.Info in it.
+			ctx := newContextWithHeaders(session.Context(), message.Headers)
+			ctx = c.obsrecv.StartMetricsOp(ctx)
 			attrs := attribute.NewSet(
 				attribute.String(attrInstanceName, c.id.String()),
 				attribute.String(attrPartition, strconv.Itoa(int(claim.Partition()))),
@@ -674,7 +681,7 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 			c.headerExtractor.extractHeadersMetrics(metrics, message)
 
 			dataPointCount := metrics.DataPointCount()
-			err = c.nextConsumer.ConsumeMetrics(session.Context(), metrics)
+			err = c.nextConsumer.ConsumeMetrics(ctx, metrics)
 			c.obsrecv.EndMetricsOp(ctx, c.unmarshaler.Encoding(), dataPointCount, err)
 			if err != nil {
 				if errorRequiresBackoff(err) && c.backOff != nil {
@@ -767,7 +774,10 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 				session.MarkMessage(message, "")
 			}
 
-			ctx := c.obsrecv.StartLogsOp(session.Context())
+			// If the Kafka exporter has propagated headers in the message,
+			// create a new context with client.Info in it.
+			ctx := newContextWithHeaders(session.Context(), message.Headers)
+			ctx = c.obsrecv.StartLogsOp(ctx)
 			attrs := attribute.NewSet(
 				attribute.String(attrInstanceName, c.id.String()),
 				attribute.String(attrPartition, strconv.Itoa(int(claim.Partition()))),
@@ -787,7 +797,7 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 			}
 			c.headerExtractor.extractHeadersLogs(logs, message)
 			logRecordCount := logs.LogRecordCount()
-			err = c.nextConsumer.ConsumeLogs(session.Context(), logs)
+			err = c.nextConsumer.ConsumeLogs(ctx, logs)
 			c.obsrecv.EndLogsOp(ctx, c.unmarshaler.Encoding(), logRecordCount, err)
 			if err != nil {
 				if errorRequiresBackoff(err) && c.backOff != nil {
@@ -892,3 +902,18 @@ func encodingToComponentID(encoding string) (*component.ID, error) {
 func errorRequiresBackoff(err error) bool {
 	return err.Error() == errMemoryLimiterDataRefused.Error()
 }
+
+func newContextWithHeaders(ctx context.Context,
+	headers []*sarama.RecordHeader,
+) context.Context {
+	if len(headers) == 0 {
+		return ctx
+	}
+	m := make(map[string][]string, len(headers))
+	for _, header := range headers {
+		key := string(header.Key)
+		value := string(header.Value)
+		m[key] = append(m[key], value)
+	}
+	return client.NewContext(ctx, client.Info{Metadata: client.NewMetadata(m)})
+}

receiver/kafkareceiver/kafka_receiver.go

@@ -7,15 +7,18 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/IBM/sarama"
 	"github.com/cenkalti/backoff/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -111,11 +114,22 @@ func TestTracesConsumerGroupHandler(t *testing.T) {
 
 	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{ReceiverCreateSettings: receivertest.NewNopSettings(metadata.Type)})
 	require.NoError(t, err)
+	var called atomic.Bool
 	c := tracesConsumerGroupHandler{
-		unmarshaler:      newPdataTracesUnmarshaler(&ptrace.ProtoUnmarshaler{}, defaultEncoding),
-		logger:           zap.NewNop(),
-		ready:            make(chan bool),
-		nextConsumer:     consumertest.NewNop(),
+		unmarshaler: newPdataTracesUnmarshaler(&ptrace.ProtoUnmarshaler{}, defaultEncoding),
+		logger:      zap.NewNop(),
+		ready:       make(chan bool),
+		nextConsumer: func() consumer.Traces {
+			c, err := consumer.NewTraces(func(ctx context.Context, _ ptrace.Traces) error {
+				defer called.CompareAndSwap(false, true)
+				info := client.FromContext(ctx)
+				assert.Equal(t, []string{"abcdefg"}, info.Metadata.Get("x-tenant-id"))
+				assert.Equal(t, []string{"1234", "5678"}, info.Metadata.Get("x-request-ids"))
+				return nil
+			})
+			require.NoError(t, err)
+			return c
+		}(),
 		obsrecv:          obsrecv,
 		headerExtractor:  &nopHeaderExtractor{},
 		telemetryBuilder: telemetryBuilder,
@@ -141,9 +155,25 @@ func TestTracesConsumerGroupHandler(t *testing.T) {
 		wg.Done()
 	}()
 
-	groupClaim.messageChan <- &sarama.ConsumerMessage{}
+	groupClaim.messageChan <- &sarama.ConsumerMessage{
+		Headers: []*sarama.RecordHeader{
+			{
+				Key:   []byte("x-tenant-id"),
+				Value: []byte("abcdefg"),
+			},
+			{
+				Key:   []byte("x-request-ids"),
+				Value: []byte("1234"),
+			},
+			{
+				Key:   []byte("x-request-ids"),
+				Value: []byte("5678"),
+			},
+		},
+	}
 	close(groupClaim.messageChan)
 	wg.Wait()
+	assert.True(t, called.Load()) // Ensure nextConsumer was called.
 }
 
 func TestTracesConsumerGroupHandler_session_done(t *testing.T) {
@@ -410,11 +440,22 @@ func TestMetricsConsumerGroupHandler(t *testing.T) {
 
 	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{ReceiverCreateSettings: receivertest.NewNopSettings(metadata.Type)})
 	require.NoError(t, err)
+	var called atomic.Bool
 	c := metricsConsumerGroupHandler{
-		unmarshaler:      newPdataMetricsUnmarshaler(&pmetric.ProtoUnmarshaler{}, defaultEncoding),
-		logger:           zap.NewNop(),
-		ready:            make(chan bool),
-		nextConsumer:     consumertest.NewNop(),
+		unmarshaler: newPdataMetricsUnmarshaler(&pmetric.ProtoUnmarshaler{}, defaultEncoding),
+		logger:      zap.NewNop(),
+		ready:       make(chan bool),
+		nextConsumer: func() consumer.Metrics {
+			c, err := consumer.NewMetrics(func(ctx context.Context, _ pmetric.Metrics) error {
+				defer called.CompareAndSwap(false, true)
+				info := client.FromContext(ctx)
+				assert.Equal(t, []string{"abcdefg"}, info.Metadata.Get("x-tenant-id"))
+				assert.Equal(t, []string{"1234", "5678"}, info.Metadata.Get("x-request-ids"))
+				return nil
+			})
+			require.NoError(t, err)
+			return c
+		}(),
 		obsrecv:          obsrecv,
 		headerExtractor:  &nopHeaderExtractor{},
 		telemetryBuilder: telemetryBuilder,
@@ -440,9 +481,25 @@ func TestMetricsConsumerGroupHandler(t *testing.T) {
 		wg.Done()
 	}()
 
-	groupClaim.messageChan <- &sarama.ConsumerMessage{}
+	groupClaim.messageChan <- &sarama.ConsumerMessage{
+		Headers: []*sarama.RecordHeader{
+			{
+				Key:   []byte("x-tenant-id"),
+				Value: []byte("abcdefg"),
+			},
+			{
+				Key:   []byte("x-request-ids"),
+				Value: []byte("1234"),
+			},
+			{
+				Key:   []byte("x-request-ids"),
+				Value: []byte("5678"),
+			},
+		},
+	}
 	close(groupClaim.messageChan)
 	wg.Wait()
+	assert.True(t, called.Load()) // Ensure nextConsumer was called.
 }
 
 func TestMetricsConsumerGroupHandler_session_done(t *testing.T) {
@@ -722,11 +779,22 @@ func TestLogsConsumerGroupHandler(t *testing.T) {
 
 	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{ReceiverCreateSettings: receivertest.NewNopSettings(metadata.Type)})
 	require.NoError(t, err)
+	var called atomic.Bool
 	c := logsConsumerGroupHandler{
-		unmarshaler:      newPdataLogsUnmarshaler(&plog.ProtoUnmarshaler{}, defaultEncoding),
-		logger:           zap.NewNop(),
-		ready:            make(chan bool),
-		nextConsumer:     consumertest.NewNop(),
+		unmarshaler: newPdataLogsUnmarshaler(&plog.ProtoUnmarshaler{}, defaultEncoding),
+		logger:      zap.NewNop(),
+		ready:       make(chan bool),
+		nextConsumer: func() consumer.Logs {
+			c, err := consumer.NewLogs(func(ctx context.Context, _ plog.Logs) error {
+				defer called.CompareAndSwap(false, true)
+				info := client.FromContext(ctx)
+				assert.Equal(t, []string{"abcdefg"}, info.Metadata.Get("x-tenant-id"))
+				assert.Equal(t, []string{"1234", "5678"}, info.Metadata.Get("x-request-ids"))
+				return nil
+			})
+			require.NoError(t, err)
+			return c
+		}(),
 		obsrecv:          obsrecv,
 		headerExtractor:  &nopHeaderExtractor{},
 		telemetryBuilder: telemetryBuilder,
@@ -752,9 +820,25 @@ func TestLogsConsumerGroupHandler(t *testing.T) {
 		wg.Done()
 	}()
 
-	groupClaim.messageChan <- &sarama.ConsumerMessage{}
+	groupClaim.messageChan <- &sarama.ConsumerMessage{
+		Headers: []*sarama.RecordHeader{
+			{
+				Key:   []byte("x-tenant-id"),
+				Value: []byte("abcdefg"),
+			},
+			{
+				Key:   []byte("x-request-ids"),
+				Value: []byte("1234"),
+			},
+			{
+				Key:   []byte("x-request-ids"),
+				Value: []byte("5678"),
+			},
+		},
+	}
 	close(groupClaim.messageChan)
 	wg.Wait()
+	assert.True(t, called.Load()) // Ensure nextConsumer was called.
 }
 
 func TestLogsConsumerGroupHandler_session_done(t *testing.T) {
@@ -1268,3 +1352,74 @@ func (c *nopNoUnmarshalerComponent) Start(_ context.Context, _ component.Host) e
 func (c *nopNoUnmarshalerComponent) Shutdown(_ context.Context) error {
 	return nil
 }
+
+func Test_newContextWithHeaders(t *testing.T) {
+	type args struct {
+		ctx     context.Context
+		headers []*sarama.RecordHeader
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string][]string
+	}{
+		{
+			name: "no headers",
+			args: args{
+				ctx:     context.Background(),
+				headers: []*sarama.RecordHeader{},
+			},
+			want: map[string][]string{},
+		},
+		{
+			name: "single header",
+			args: args{
+				ctx: context.Background(),
+				headers: []*sarama.RecordHeader{
+					{Key: []byte("key1"), Value: []byte("value1")},
+				},
+			},
+			want: map[string][]string{
+				"key1": {"value1"},
+			},
+		},
+		{
+			name: "multiple headers",
+			args: args{
+				ctx: context.Background(),
+				headers: []*sarama.RecordHeader{
+					{Key: []byte("key1"), Value: []byte("value1")},
+					{Key: []byte("key2"), Value: []byte("value2")},
+				},
+			},
+			want: map[string][]string{
+				"key1": {"value1"},
+				"key2": {"value2"},
+			},
+		},
+		{
+			name: "duplicate keys",
+			args: args{
+				ctx: context.Background(),
+				headers: []*sarama.RecordHeader{
+					{Key: []byte("key1"), Value: []byte("value1")},
+					{Key: []byte("key1"), Value: []byte("value2")},
+				},
+			},
+			want: map[string][]string{
+				"key1": {"value1", "value2"},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := newContextWithHeaders(tt.args.ctx, tt.args.headers)
+			clientInfo := client.FromContext(ctx)
+			for k, wantVal := range tt.want {
+				val := clientInfo.Metadata.Get(k)
+				assert.Equal(t, wantVal, val)
+			}
+		})
+	}
+}