kafkaexporter: Opt-in to use franz-go client (open-telemetry#40364)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
Adds the option to use the franz-go client in the `kafkaexporter`. The
main rationale for this change is to allow users to opt-in to use a
more performant Kafka client, as the default Sarama client has been
known to have performance issues in high-throughput scenarios.

The change _should_ be backward compatible, and significantly increases
the production performance of the `kafkaexporter` when in use.

```
$ benchstat sarama.txt franz-go.txt
goos: darwin
goarch: arm64
pkg: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter
cpu: Apple M2 Pro
           │ sarama.txt  │             franz-go.txt             │
           │   sec/op    │    sec/op     vs base                │
Logs-12      55.56µ ± 5%   33.40µ ±  4%  -39.88% (p=0.000 n=10)
Metrics-12   60.69µ ± 6%   39.34µ ± 17%  -35.18% (p=0.001 n=10)
Traces-12    57.09µ ± 9%   35.71µ ± 13%  -37.44% (p=0.000 n=10)
geomean      57.74µ        36.07µ        -37.53%

           │  sarama.txt  │             franz-go.txt              │
           │     B/op     │     B/op       vs base                │
Logs-12      5.792Ki ± 0%   2.207Ki ±  1%  -61.90% (p=0.000 n=10)
Metrics-12   9.545Ki ± 0%   4.339Ki ± 16%  -54.54% (p=0.000 n=10)
Traces-12    7.329Ki ± 0%   2.967Ki ±  5%  -59.52% (p=0.000 n=10)
geomean      7.400Ki        3.051Ki        -58.76%

           │ sarama.txt │            franz-go.txt            │
           │ allocs/op  │ allocs/op   vs base                │
Logs-12      55.00 ± 0%   29.00 ± 0%  -47.27% (p=0.000 n=10)
Metrics-12   58.00 ± 0%   32.00 ± 0%  -44.83% (p=0.000 n=10)
Traces-12    55.00 ± 0%   29.00 ± 0%  -47.27% (p=0.000 n=10)
geomean      55.98        29.97       -46.47%
```

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Improves the exporter's production performance.

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Unit tested, benchmarked against a local Kafka broker.

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

Updated the Readme.

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

---------

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

Author: marclop

.chloggen/f_add-kafkaexporter-franz-go-client.yaml

@@ -0,0 +1,31 @@
+# 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: 'kafkaexporter'
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add an Alpha feature gate `exporter.kafkaexporter.UseFranzGoClient` to use franz-go in the Kafka exporter for better performance."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40364]
+
+# (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: |
+  This change adds an experimental opt-in support to use the franz-go client in the Kafka exporter.
+  The franz-go client is a high-performance Kafka client that can improve the performance of the Kafka exporter.
+  The default client remains sarama, which is used by the Kafka receiver and other components.
+  Enable the franz-go client by setting the `exporter.kafkaexporter.UseFranzGo` feature gate.
+
+# 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]

exporter/kafkaexporter/README.md

@@ -20,9 +20,15 @@ processors for higher throughput and resiliency. Message payload encoding is con
 
 ## Configuration settings
 
+> [!NOTE]
+> You can opt-in to use [`franz-go`](https://github.com/twmb/franz-go) client by enabling the feature gate
+> `exporter.kafkaexporter.UseFranzGo` when you run the OpenTelemetry Collector. See the following page
+> for more details: [Feature Gates](https://github.com/open-telemetry/opentelemetry-collector/tree/main/featuregate#controlling-gates)
+
 There are no required settings.
 
 The following settings can be optionally configured:
+
 - `brokers` (default = localhost:9092): The list of kafka brokers.
 - `protocol_version` (default = 2.1.0): Kafka protocol version.
 - `resolve_canonical_bootstrap_servers_only` (default = false): Whether to resolve then reverse-lookup broker IPs during startup.

exporter/kafkaexporter/generated_package_test.go

@@ -16,6 +16,8 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.127.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin v0.127.0
 	github.com/stretchr/testify v1.10.0
+	github.com/twmb/franz-go v1.18.1
+	github.com/twmb/franz-go/pkg/kfake v0.0.0-20250320172111-35ab5e5f5327
 	go.opentelemetry.io/collector/client v1.33.1-0.20250603105141-605011a1fea8
 	go.opentelemetry.io/collector/component v1.33.1-0.20250603105141-605011a1fea8
 	go.opentelemetry.io/collector/component/componenttest v0.127.1-0.20250603105141-605011a1fea8
@@ -26,6 +28,7 @@ require (
 	go.opentelemetry.io/collector/consumer/consumererror v0.127.1-0.20250603105141-605011a1fea8
 	go.opentelemetry.io/collector/exporter v0.127.1-0.20250603105141-605011a1fea8
 	go.opentelemetry.io/collector/exporter/exportertest v0.127.1-0.20250603105141-605011a1fea8
+	go.opentelemetry.io/collector/featuregate v1.33.1-0.20250603105141-605011a1fea8
 	go.opentelemetry.io/collector/pdata v1.33.1-0.20250603105141-605011a1fea8
 	go.opentelemetry.io/collector/pdata/testdata v0.127.1-0.20250603105141-605011a1fea8
 	go.uber.org/goleak v1.3.0
@@ -87,6 +90,9 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.11.2 // indirect
+	github.com/twmb/franz-go/pkg/sasl/kerberos v1.1.0 // indirect
+	github.com/twmb/franz-go/plugin/kzap v1.1.2 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
@@ -99,7 +105,6 @@ require (
 	go.opentelemetry.io/collector/exporter/xexporter v0.127.1-0.20250603105141-605011a1fea8 // indirect
 	go.opentelemetry.io/collector/extension v1.33.1-0.20250603105141-605011a1fea8 // indirect
 	go.opentelemetry.io/collector/extension/xextension v0.127.1-0.20250603105141-605011a1fea8 // indirect
-	go.opentelemetry.io/collector/featuregate v1.33.1-0.20250603105141-605011a1fea8 // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.127.1-0.20250603105141-605011a1fea8 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.127.1-0.20250603105141-605011a1fea8 // indirect
 	go.opentelemetry.io/collector/pipeline v0.127.1-0.20250603105141-605011a1fea8 // indirect

exporter/kafkaexporter/go.mod

@@ -0,0 +1,6 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+// Package kafkaclient provides implementations of Kafka producers using
+// different client libraries (Sarama, Franz-go).
+package kafkaclient // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient"

exporter/kafkaexporter/go.sum

@@ -0,0 +1,72 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kafkaclient // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient"
+
+import (
+	"context"
+	"errors"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+)
+
+// FranzSyncProducer is a wrapper around the franz-go client that implements
+// the Producer interface. Allowing us to use the franz-go client while
+// maintaining compatibility with the existing Kafka exporter code.
+type FranzSyncProducer struct {
+	client       *kgo.Client
+	metadataKeys []string
+}
+
+// NewFranzSyncProducer Franz-go producer from a kgo.Client and a Messenger.
+func NewFranzSyncProducer(client *kgo.Client,
+	metadataKeys []string,
+) *FranzSyncProducer {
+	return &FranzSyncProducer{
+		client:       client,
+		metadataKeys: metadataKeys,
+	}
+}
+
+// ExportData sends a batch of messages to Kafka
+func (p *FranzSyncProducer) ExportData(ctx context.Context, msgs Messages) error {
+	messages := makeFranzMessages(msgs)
+	setMessageHeaders(ctx, messages, p.metadataKeys,
+		func(key string, value []byte) kgo.RecordHeader {
+			return kgo.RecordHeader{Key: key, Value: value}
+		},
+		func(m *kgo.Record) []kgo.RecordHeader { return m.Headers },
+		func(m *kgo.Record, h []kgo.RecordHeader) { m.Headers = h },
+	)
+	result := p.client.ProduceSync(ctx, messages...)
+	var errs []error
+	for _, r := range result {
+		if r.Err != nil {
+			errs = append(errs, r.Err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// Close shuts down the producer and flushes any remaining messages.
+func (p *FranzSyncProducer) Close() error {
+	p.client.Close()
+	return nil
+}
+
+func makeFranzMessages(messages Messages) []*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}
+			if message.Key != nil {
+				msg.Key = message.Key
+			}
+			if message.Value != nil {
+				msg.Value = message.Value
+			}
+			msgs = append(msgs, msg)
+		}
+	}
+	return msgs
+}

exporter/kafkaexporter/internal/kafkaclient/doc.go

@@ -0,0 +1,72 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kafkaclient // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/client"
+)
+
+// metadataToHeaders converts metadata from the context into a slice of headers using the provided header constructor.
+// This function is generic and can be used for both Sarama and Franz-go header types.
+func metadataToHeaders[H any](ctx context.Context, keys []string,
+	makeHeader func(key string, value []byte) H,
+) []H {
+	if len(keys) == 0 {
+		return nil
+	}
+	info := client.FromContext(ctx)
+	headers := make([]H, 0, len(keys))
+	for _, key := range keys {
+		valueSlice := info.Metadata.Get(key)
+		for _, v := range valueSlice {
+			headers = append(headers, makeHeader(key, []byte(v)))
+		}
+	}
+	return headers
+}
+
+// setMessageHeaders is a generic helper for setting headers on a slice of messages.
+// - messages: the messages to set headers on
+// - ctx: context for extracting metadata
+// - metadataKeys: which metadata keys to extract
+// - makeHeader: constructs the header type for the target client (Sarama/Franz-go)
+// - getHeaders: gets the headers from a message
+// - setHeaders: sets the headers on a message
+// Usage example (Sarama):
+//
+//	setMessageHeaders(ctx, allMessages, keys, makeHeader, getHeaders, setHeaders)
+func setMessageHeaders[M any, H any](ctx context.Context,
+	messages []M,
+	metadataKeys []string,
+	makeHeader func(key string, value []byte) H,
+	getHeaders func(M) []H,
+	setHeadersFunc func(M, []H),
+) {
+	setHeaders(
+		messages,
+		metadataToHeaders(ctx, metadataKeys, makeHeader),
+		getHeaders,
+		setHeadersFunc,
+	)
+}
+
+// setHeaders sets or appends headers on each message in messages using the provided get/set functions.
+func setHeaders[M any, H any](messages []M, headers []H,
+	getHeaders func(M) []H,
+	setHeaders func(M, []H),
+) {
+	if len(headers) == 0 || len(messages) == 0 {
+		return
+	}
+	for i := range messages {
+		h := getHeaders(messages[i])
+		if len(h) == 0 {
+			setHeaders(messages[i], headers)
+		} else {
+			setHeaders(messages[i], append(h, headers...))
+		}
+	}
+}

exporter/kafkaexporter/internal/kafkaclient/franzgo.go

@@ -0,0 +1,21 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kafkaclient // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient"
+
+import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/marshaler"
+
+// Messages is a collection of messages (with count) to be sent to Kafka.
+type Messages struct {
+	// Count is the total number of messages across all topics.
+	// Populating this field allows the downstream method to preallocate
+	// the slice of messages, which can improve performance.
+	Count         int
+	TopicMessages []TopicMessages
+}
+
+// TopicMessages represents a collection of messages for a specific topic.
+type TopicMessages struct {
+	Topic    string
+	Messages []marshaler.Message
+}