kafkaexporter: Propagate metadata keys as headers

Adds a new config option specifying a list of metadata keys that should
be propagated as Kafka message headers.

Closes open-telemetry#39130

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

Author: marclop

exporter/kafkaexporter/README.md

@@ -26,6 +26,7 @@ The following settings can be optionally configured:
 - `client_id` (default = "otel-collector"): The client ID to configure the Kafka client with. The client ID will be used for all produce requests.
 - `topic` (default = otlp_spans for traces, otlp_metrics for metrics, otlp_logs for logs): The name of the default kafka topic to export to. See [Destination Topic](#destination-topic) below for more details.
 - `topic_from_attribute` (default = ""): Specify the resource attribute whose value should be used as the message's topic. See [Destination Topic](#destination-topic) below for more details. 
+- `include_metadata_keys` (default = []): Specifies a list of metadata keys to propagate as Kafka message headers. If one or more keys aren't found in the metadata, they are ignored. Additionally, if the metadata key contains more than one item, it won't be propagated.
 - `encoding` (default = otlp_proto): The encoding of the traces sent to kafka. All available encodings:
   - `otlp_proto`: payload is Protobuf serialized from `ExportTraceServiceRequest` if set as a traces exporter or `ExportMetricsServiceRequest` for metrics or `ExportLogsServiceRequest` for logs.
   - `otlp_json`:  payload is JSON serialized from `ExportTraceServiceRequest` if set as a traces exporter or `ExportMetricsServiceRequest` for metrics or `ExportLogsServiceRequest` for logs. 

exporter/kafkaexporter/config.go

@@ -24,6 +24,9 @@ type Config struct {
 	// The name of the kafka topic to export to (default otlp_spans for traces, otlp_metrics for metrics)
 	Topic string `mapstructure:"topic"`
 
+	// IncludeMetadataKeys indicates the receiver's client metadata keys to propagate as Kafka message headers.
+	IncludeMetadataKeys []string `mapstructure:"include_metadata_keys"`
+
 	// TopicFromAttribute is the name of the attribute to use as the topic name.
 	TopicFromAttribute string `mapstructure:"topic_from_attribute"`
 

exporter/kafkaexporter/go.mod

@@ -15,6 +15,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

exporter/kafkaexporter/go.sum

@@ -9,6 +9,7 @@ import (
 	"fmt"
 
 	"github.com/IBM/sarama"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/exporter"
@@ -46,6 +47,9 @@ func (e *kafkaTracesProducer) tracesPusher(ctx context.Context, td ptrace.Traces
 	if err != nil {
 		return consumererror.NewPermanent(err)
 	}
+	messagesWithHeaders(messages,
+		metadataToHeaders(ctx, e.cfg.IncludeMetadataKeys),
+	)
 	err = e.producer.SendMessages(messages)
 	if err != nil {
 		var prodErr sarama.ProducerErrors
@@ -104,6 +108,9 @@ func (e *kafkaMetricsProducer) metricsDataPusher(ctx context.Context, md pmetric
 	if err != nil {
 		return consumererror.NewPermanent(err)
 	}
+	messagesWithHeaders(messages,
+		metadataToHeaders(ctx, e.cfg.IncludeMetadataKeys),
+	)
 	err = e.producer.SendMessages(messages)
 	if err != nil {
 		var prodErr sarama.ProducerErrors
@@ -162,6 +169,9 @@ func (e *kafkaLogsProducer) logsDataPusher(ctx context.Context, ld plog.Logs) er
 	if err != nil {
 		return consumererror.NewPermanent(err)
 	}
+	messagesWithHeaders(messages,
+		metadataToHeaders(ctx, e.cfg.IncludeMetadataKeys),
+	)
 	err = e.producer.SendMessages(messages)
 	if err != nil {
 		var prodErr sarama.ProducerErrors
@@ -287,3 +297,35 @@ func encodingToComponentID(encoding string) (*component.ID, error) {
 	id := component.NewID(componentType)
 	return &id, nil
 }
+
+func messagesWithHeaders(msg []*sarama.ProducerMessage, h []sarama.RecordHeader) {
+	if len(h) == 0 || len(msg) == 0 {
+		return
+	}
+	for i := range msg {
+		if len(msg[i].Headers) == 0 {
+			msg[i].Headers = h
+			continue
+		}
+		msg[i].Headers = append(msg[i].Headers, h...)
+	}
+}
+
+func metadataToHeaders(ctx context.Context, keys []string) []sarama.RecordHeader {
+	if len(keys) == 0 {
+		return nil
+	}
+	info := client.FromContext(ctx)
+	headers := make([]sarama.RecordHeader, 0, len(keys))
+	for _, key := range keys {
+		v := info.Metadata.Get(key)
+		if len(v) != 1 {
+			continue
+		}
+		headers = append(headers, sarama.RecordHeader{
+			Key:   []byte(key),
+			Value: []byte(v[0]),
+		})
+	}
+	return headers
+}

exporter/kafkaexporter/kafka_exporter.go

@@ -12,6 +12,7 @@ import (
 	"github.com/IBM/sarama/mocks"
 	"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/exporter/exportertest"
@@ -134,19 +135,52 @@ func TestTracesPusher_attr(t *testing.T) {
 }
 
 func TestTracesPusher_ctx(t *testing.T) {
-	c := sarama.NewConfig()
-	producer := mocks.NewSyncProducer(t, c)
-	producer.ExpectSendMessageAndSucceed()
+	t.Run("WithTopic", func(t *testing.T) {
+		c := sarama.NewConfig()
+		producer := mocks.NewSyncProducer(t, c)
+		producer.ExpectSendMessageAndSucceed()
+
+		p := kafkaTracesProducer{
+			producer:  producer,
+			marshaler: newPdataTracesMarshaler(&ptrace.ProtoMarshaler{}, defaultEncoding, false),
+		}
+		t.Cleanup(func() {
+			require.NoError(t, p.Close(context.Background()))
+		})
+		err := p.tracesPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateTraces(2))
+		require.NoError(t, err)
+	})
+	t.Run("WithMetadata", func(t *testing.T) {
+		c := sarama.NewConfig()
+		producer := mocks.NewSyncProducer(t, c)
+		producer.ExpectSendMessageWithMessageCheckerFunctionAndSucceed(func(pm *sarama.ProducerMessage) error {
+			assert.Equal(t, []sarama.RecordHeader{
+				{Key: []byte("x-tenant-id"), Value: []byte("my_tenant_id")},
+				{Key: []byte("x-request-id"), Value: []byte("123456789")},
+			}, pm.Headers)
+			return nil
+		})
 
-	p := kafkaTracesProducer{
-		producer:  producer,
-		marshaler: newPdataTracesMarshaler(&ptrace.ProtoMarshaler{}, defaultEncoding, false),
-	}
-	t.Cleanup(func() {
-		require.NoError(t, p.Close(context.Background()))
+		p := kafkaTracesProducer{
+			cfg: Config{
+				IncludeMetadataKeys: []string{"x-tenant-id", "x-request-id"},
+			},
+			producer:  producer,
+			marshaler: newPdataTracesMarshaler(&ptrace.ProtoMarshaler{}, defaultEncoding, false),
+		}
+		t.Cleanup(func() {
+			require.NoError(t, p.Close(context.Background()))
+		})
+		ctx := client.NewContext(context.Background(), client.Info{
+			Metadata: client.NewMetadata(map[string][]string{
+				"x-tenant-id":    {"my_tenant_id"},
+				"x-request-id":   {"123456789"},
+				"discarded-meta": {"my-meta"}, // This will be ignored.
+			}),
+		})
+		err := p.tracesPusher(ctx, testdata.GenerateTraces(10))
+		require.NoError(t, err)
 	})
-	err := p.tracesPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateTraces(2))
-	require.NoError(t, err)
 }
 
 func TestTracesPusher_err(t *testing.T) {
@@ -215,19 +249,52 @@ func TestMetricsDataPusher_attr(t *testing.T) {
 }
 
 func TestMetricsDataPusher_ctx(t *testing.T) {
-	c := sarama.NewConfig()
-	producer := mocks.NewSyncProducer(t, c)
-	producer.ExpectSendMessageAndSucceed()
+	t.Run("WithTopic", func(t *testing.T) {
+		c := sarama.NewConfig()
+		producer := mocks.NewSyncProducer(t, c)
+		producer.ExpectSendMessageAndSucceed()
+
+		p := kafkaMetricsProducer{
+			producer:  producer,
+			marshaler: newPdataMetricsMarshaler(&pmetric.ProtoMarshaler{}, defaultEncoding, false),
+		}
+		t.Cleanup(func() {
+			require.NoError(t, p.Close(context.Background()))
+		})
+		err := p.metricsDataPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateMetrics(2))
+		require.NoError(t, err)
+	})
+	t.Run("WithMetadata", func(t *testing.T) {
+		c := sarama.NewConfig()
+		producer := mocks.NewSyncProducer(t, c)
+		producer.ExpectSendMessageWithMessageCheckerFunctionAndSucceed(func(pm *sarama.ProducerMessage) error {
+			assert.Equal(t, []sarama.RecordHeader{
+				{Key: []byte("x-tenant-id"), Value: []byte("anoter_tenant_id")},
+				{Key: []byte("x-request-id"), Value: []byte("987654321")},
+			}, pm.Headers)
+			return nil
+		})
 
-	p := kafkaMetricsProducer{
-		producer:  producer,
-		marshaler: newPdataMetricsMarshaler(&pmetric.ProtoMarshaler{}, defaultEncoding, false),
-	}
-	t.Cleanup(func() {
-		require.NoError(t, p.Close(context.Background()))
+		p := kafkaMetricsProducer{
+			cfg: Config{
+				IncludeMetadataKeys: []string{"x-tenant-id", "x-request-id"},
+			},
+			producer:  producer,
+			marshaler: newPdataMetricsMarshaler(&pmetric.ProtoMarshaler{}, defaultEncoding, false),
+		}
+		t.Cleanup(func() {
+			require.NoError(t, p.Close(context.Background()))
+		})
+		ctx := client.NewContext(context.Background(), client.Info{
+			Metadata: client.NewMetadata(map[string][]string{
+				"x-tenant-id":    {"anoter_tenant_id"},
+				"x-request-id":   {"987654321"},
+				"discarded-meta": {"my-meta"}, // This will be ignored.
+			}),
+		})
+		err := p.metricsDataPusher(ctx, testdata.GenerateMetrics(5))
+		require.NoError(t, err)
 	})
-	err := p.metricsDataPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateMetrics(2))
-	require.NoError(t, err)
 }
 
 func TestMetricsDataPusher_err(t *testing.T) {
@@ -296,19 +363,52 @@ func TestLogsDataPusher_attr(t *testing.T) {
 }
 
 func TestLogsDataPusher_ctx(t *testing.T) {
-	c := sarama.NewConfig()
-	producer := mocks.NewSyncProducer(t, c)
-	producer.ExpectSendMessageAndSucceed()
+	t.Run("WithTopic", func(t *testing.T) {
+		c := sarama.NewConfig()
+		producer := mocks.NewSyncProducer(t, c)
+		producer.ExpectSendMessageAndSucceed()
+
+		p := kafkaLogsProducer{
+			producer:  producer,
+			marshaler: newPdataLogsMarshaler(&plog.ProtoMarshaler{}, defaultEncoding, false),
+		}
+		t.Cleanup(func() {
+			require.NoError(t, p.Close(context.Background()))
+		})
+		err := p.logsDataPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateLogs(1))
+		require.NoError(t, err)
+	})
+	t.Run("WithMetadata", func(t *testing.T) {
+		c := sarama.NewConfig()
+		producer := mocks.NewSyncProducer(t, c)
+		producer.ExpectSendMessageWithMessageCheckerFunctionAndSucceed(func(pm *sarama.ProducerMessage) error {
+			assert.Equal(t, []sarama.RecordHeader{
+				{Key: []byte("x-tenant-id"), Value: []byte("yet_another_tenant_id")},
+				{Key: []byte("x-request-id"), Value: []byte("01234")},
+			}, pm.Headers)
+			return nil
+		})
 
-	p := kafkaLogsProducer{
-		producer:  producer,
-		marshaler: newPdataLogsMarshaler(&plog.ProtoMarshaler{}, defaultEncoding, false),
-	}
-	t.Cleanup(func() {
-		require.NoError(t, p.Close(context.Background()))
+		p := kafkaLogsProducer{
+			cfg: Config{
+				IncludeMetadataKeys: []string{"x-tenant-id", "x-request-id"},
+			},
+			producer:  producer,
+			marshaler: newPdataLogsMarshaler(&plog.ProtoMarshaler{}, defaultEncoding, false),
+		}
+		t.Cleanup(func() {
+			require.NoError(t, p.Close(context.Background()))
+		})
+		ctx := client.NewContext(context.Background(), client.Info{
+			Metadata: client.NewMetadata(map[string][]string{
+				"x-tenant-id":    {"yet_another_tenant_id"},
+				"x-request-id":   {"01234"},
+				"discarded-meta": {"my-meta"}, // This will be ignored.
+			}),
+		})
+		err := p.logsDataPusher(ctx, testdata.GenerateLogs(15))
+		require.NoError(t, err)
 	})
-	err := p.logsDataPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateLogs(1))
-	require.NoError(t, err)
 }
 
 func TestLogsDataPusher_err(t *testing.T) {