[exporter/kafka] Add record_headers config to set static headers on outgoing records (#47201)

## Description

Resolves #47193 

This PR adds a new `record_headers` configuration option to the Kafka
exporter

**Changes made:**
* added `record_headers` to `Config` struct
* updated `makeFranzMessages` in the `franz-go` client wrapper to
iterate over the configured map and append them as `kgo.RecordHeader` to
the outgoing messages
* added `TestMakeFranzMessages_RecordHeaders` to verify headers are
successfully attached, bringing test suite to 147 passing tests
* documented the new field and provided an example in the `README.md`
* added `.chloggen/fix-47193.yaml`

## Testing

- [x] added `TestMakeFranzMessages_RecordHeaders`
- [x] all 147 tests passed locally

---------

Signed-off-by: singhvibhanshu <singhvibhanshu@hotmail.com>

Author: singhvibhanshu

.chloggen/fix_47193.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. receiver/filelog)
+component: exporter/kafka
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `record_headers` configuration option to set static headers on outgoing records
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47193]
+
+# (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: []

exporter/kafkaexporter/README.md

@@ -48,6 +48,7 @@ The following settings can be optionally configured:
   - `topic_from_metadata_key` (default = ""): The name of the metadata key whose value should be used as the message's topic. Useful to dynamically produce to topics based on request inputs. It takes precedence over `topic_from_attribute` and `topic` settings.
 - `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. When `sending_queue::batch` is enabled, `sending_queue::batch::partition::metadata_keys` must be configured and include all values configured in `include_metadata_keys`.
+- `record_headers` (default = {}): Specifies a map of key/value pairs to set as static headers on every outgoing Kafka record.
 - `partition_traces_by_id` (default = false): configures the exporter to include the trace ID as the message key in trace messages sent to kafka. *Please note:* this setting does not have any effect on Jaeger encoding exporters since Jaeger exporters include trace ID as the message key by default.
 - `partition_metrics_by_resource_attributes` (default = false)  configures the exporter to include the hash of sorted resource attributes as the message partitioning key in metric messages sent to kafka.
 - `partition_logs_by_resource_attributes` (default = false)  configures the exporter to include the hash of sorted resource attributes as the message partitioning key in log messages sent to kafka.
@@ -158,6 +159,9 @@ exporters:
   kafka:
     brokers:
       - localhost:9092
+    record_headers:
+      my-custom-header: "my-custom-value"
+      another-header: "another-value"
 ```
 
 ## Destination Topic

exporter/kafkaexporter/config.go

@@ -10,6 +10,7 @@ import (
 	"slices"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configoptional"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/confmap"
@@ -142,6 +143,9 @@ type Config struct {
 	// IncludeMetadataKeys indicates the receiver's client metadata keys to propagate as Kafka message headers.
 	IncludeMetadataKeys []string `mapstructure:"include_metadata_keys"`
 
+	// RecordHeaders sets static headers on every outgoing Kafka record.
+	RecordHeaders configopaque.MapList `mapstructure:"record_headers"`
+
 	// TopicFromAttribute is the name of the attribute to use as the topic name.
 	TopicFromAttribute string `mapstructure:"topic_from_attribute"`
 

exporter/kafkaexporter/config.schema.yaml

@@ -71,6 +71,9 @@ properties:
   profiles:
     description: Profiles holds configuration about how profiles should be sent to Kafka.
     $ref: signal_config
+  record_headers:
+    description: RecordHeaders sets static headers on every outgoing Kafka record.
+    $ref: go.opentelemetry.io/collector/config/configopaque.map_list
   record_partitioner:
     description: RecordPartitioner configures how Kafka records are assigned to partitions. The default ("sarama_compatible") retains the legacy Sarama-compatible hashing behavior. Set to "sticky", "round_robin", or "least_backup" to use one of the built-in franz-go partitioners, or "extension" to delegate to a custom extension.
     $ref: record_partitioner_config

exporter/kafkaexporter/go.mod

@@ -21,6 +21,7 @@ require (
 	go.opentelemetry.io/collector/client v1.55.1-0.20260409104450-d686cf9058ce
 	go.opentelemetry.io/collector/component v1.55.1-0.20260409104450-d686cf9058ce
 	go.opentelemetry.io/collector/component/componenttest v0.149.1-0.20260409104450-d686cf9058ce
+	go.opentelemetry.io/collector/config/configopaque v1.55.1-0.20260409104450-d686cf9058ce
 	go.opentelemetry.io/collector/config/configoptional v1.55.1-0.20260409104450-d686cf9058ce
 	go.opentelemetry.io/collector/config/configretry v1.55.1-0.20260409104450-d686cf9058ce
 	go.opentelemetry.io/collector/confmap v1.55.1-0.20260409104450-d686cf9058ce
@@ -104,7 +105,6 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/collector/config/configcompression v1.55.1-0.20260409104450-d686cf9058ce // indirect
-	go.opentelemetry.io/collector/config/configopaque v1.55.1-0.20260409104450-d686cf9058ce // indirect
 	go.opentelemetry.io/collector/config/configtls v1.55.1-0.20260409104450-d686cf9058ce // indirect
 	go.opentelemetry.io/collector/consumer/consumererror/xconsumererror v0.149.1-0.20260409104450-d686cf9058ce // indirect
 	go.opentelemetry.io/collector/consumer/consumertest v0.149.1-0.20260409104450-d686cf9058ce // indirect

exporter/kafkaexporter/internal/kafkaclient/franzgo.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kgo"
+	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 )
 
@@ -45,24 +46,27 @@ func recordUserSize(r *kgo.Record) int {
 type FranzSyncProducer struct {
 	client          *kgo.Client
 	metadataKeys    []string
+	recordHeaders   configopaque.MapList
 	maxMessageBytes int
 }
 
 // NewFranzSyncProducer Franz-go producer from a kgo.Client and a Messenger.
 func NewFranzSyncProducer(client *kgo.Client,
 	metadataKeys []string,
+	recordHeaders configopaque.MapList,
 	maxMessageBytes int,
 ) *FranzSyncProducer {
 	return &FranzSyncProducer{
 		client:          client,
 		metadataKeys:    metadataKeys,
+		recordHeaders:   recordHeaders,
 		maxMessageBytes: maxMessageBytes,
 	}
 }
 
 // ExportData sends a batch of messages to Kafka
 func (p *FranzSyncProducer) ExportData(ctx context.Context, msgs Messages) error {
-	messages := makeFranzMessages(msgs)
+	messages := makeFranzMessages(msgs, p.recordHeaders)
 	setMessageHeaders(ctx, messages, p.metadataKeys)
 	result := p.client.ProduceSync(ctx, messages...)
 	var errs []error
@@ -93,18 +97,24 @@ func (p *FranzSyncProducer) Close() error {
 	return nil
 }
 
-func makeFranzMessages(messages Messages) []*kgo.Record {
+func makeFranzMessages(messages Messages, recordHeaders configopaque.MapList) []*kgo.Record {
 	msgs := make([]*kgo.Record, 0, messages.Count)
 	for _, msg := range messages.TopicMessages {
 		for _, message := range msg.Messages {
-			msg := &kgo.Record{Topic: msg.Topic}
+			record := &kgo.Record{Topic: msg.Topic}
 			if message.Key != nil {
-				msg.Key = message.Key
+				record.Key = message.Key
 			}
 			if message.Value != nil {
-				msg.Value = message.Value
+				record.Value = message.Value
 			}
-			msgs = append(msgs, msg)
+			for _, pair := range recordHeaders {
+				record.Headers = append(record.Headers, kgo.RecordHeader{
+					Key:   pair.Name,
+					Value: []byte(string(pair.Value)),
+				})
+			}
+			msgs = append(msgs, record)
 		}
 	}
 	return msgs

exporter/kafkaexporter/internal/kafkaclient/franzgo_test.go

@@ -12,6 +12,8 @@ import (
 	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kfake"
 	"github.com/twmb/franz-go/pkg/kgo"
+	"go.opentelemetry.io/collector/client"
+	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/marshaler"
@@ -33,7 +35,7 @@ func TestExportData_MessageTooLarge(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(client.Close)
 
-	producer := NewFranzSyncProducer(client, nil, maxMessageBytes)
+	producer := NewFranzSyncProducer(client, nil, nil, maxMessageBytes)
 
 	// Create a message larger than maxMessageBytes to trigger MessageTooLarge.
 	largeValue := []byte(strings.Repeat("x", maxMessageBytes*2))
@@ -64,3 +66,46 @@ func TestExportData_MessageTooLarge(t *testing.T) {
 	assert.Contains(t, err.Error(), "record size")
 	assert.Contains(t, err.Error(), "exceeds max")
 }
+
+func TestMakeFranzMessages_RecordHeaders(t *testing.T) {
+	recordHeaders := configopaque.MapList{
+		{Name: "static-key-ONLY", Value: configopaque.String("static-value")},
+		{Name: "shared-key", Value: configopaque.String("static-value-override")},
+	}
+
+	md := client.NewMetadata(map[string][]string{
+		"dynamic-key-ONLY": {"dynamic-value"},
+		"shared-key":       {"dynamic-value-wins"},
+	})
+	ctx := client.NewContext(t.Context(), client.Info{Metadata: md})
+
+	msgs := Messages{
+		Count: 1,
+		TopicMessages: []TopicMessages{{
+			Topic: "test-topic",
+			Messages: []marshaler.Message{{
+				Value: []byte("test-payload"),
+			}},
+		}},
+	}
+
+	records := makeFranzMessages(msgs, recordHeaders)
+	setMessageHeaders(ctx, records, []string{"dynamic-key-ONLY", "shared-key"})
+
+	require.Len(t, records, 1, "expected exactly 1 record")
+	record := records[0]
+
+	assert.Equal(t, "test-topic", record.Topic)
+	assert.Equal(t, []byte("test-payload"), record.Value)
+
+	require.Len(t, record.Headers, 4, "expected exactly 4 headers on the record")
+
+	headerMap := make(map[string]string)
+	for _, h := range record.Headers {
+		headerMap[h.Key] = string(h.Value)
+	}
+
+	assert.Equal(t, "static-value", headerMap["static-key-ONLY"], "static headers unique key failed")
+	assert.Equal(t, "dynamic-value", headerMap["dynamic-key-ONLY"], "dynamic headers unique key failed")
+	assert.Equal(t, "dynamic-value-wins", headerMap["shared-key"], "Precedence for common key failed")
+}

exporter/kafkaexporter/kafka_exporter.go

@@ -106,6 +106,7 @@ func (e *kafkaExporter[T]) Start(ctx context.Context, host component.Host) (err
 	}
 	e.producer = kafkaclient.NewFranzSyncProducer(producer,
 		e.cfg.IncludeMetadataKeys,
+		e.cfg.RecordHeaders,
 		e.cfg.Producer.MaxMessageBytes,
 	)
 	return nil

exporter/kafkaexporter/kafka_exporter_e2e_bench_test.go

@@ -50,7 +50,7 @@ func configureExporterBench[T any](
 	messenger, err := exp.newMessenger(componenttest.NewNopHost())
 	require.NoError(b, err)
 	exp.messenger = messenger
-	exp.producer = kafkaclient.NewFranzSyncProducer(client, cfg.IncludeMetadataKeys, cfg.Producer.MaxMessageBytes)
+	exp.producer = kafkaclient.NewFranzSyncProducer(client, cfg.IncludeMetadataKeys, cfg.RecordHeaders, cfg.Producer.MaxMessageBytes)
 
 	b.Cleanup(func() { exp.Close(b.Context()) })
 }

exporter/kafkaexporter/kafka_exporter_test.go

@@ -1299,7 +1299,7 @@ func configureExporter[T any](tb testing.TB,
 	require.NoError(tb, err, "failed to create messenger for metrics")
 
 	exp.messenger = messenger
-	exp.producer = kafkaclient.NewFranzSyncProducer(client, cfg.IncludeMetadataKeys, cfg.Producer.MaxMessageBytes)
+	exp.producer = kafkaclient.NewFranzSyncProducer(client, cfg.IncludeMetadataKeys, cfg.RecordHeaders, cfg.Producer.MaxMessageBytes)
 
 	tb.Cleanup(func() { assert.NoError(tb, exp.Close(tb.Context())) })
 	return cluster