[chore] Inject component.Host into franz-go Kafka clients (#45336)

#### Description

Inject `component.Host` into franz-go Kafka clients. This will enable us
to expose an oauth2 token source as an extension (e.g. oauth2clientauth,
or azureauth
(#45064)).

See also
#41873 (comment)

#### Link to tracking issue

N/A

#### Testing

N/A

#### Documentation

N/A

Author: axw

exporter/kafkaexporter/kafka_exporter.go

@@ -87,6 +87,7 @@ func (e *kafkaExporter[T]) Start(ctx context.Context, host component.Host) (err
 
 	producer, err := kafka.NewFranzSyncProducer(
 		ctx,
+		host,
 		e.cfg.ClientConfig,
 		e.cfg.Producer,
 		e.cfg.TimeoutSettings.Timeout,

exporter/kafkaexporter/kafka_exporter_test.go

@@ -949,7 +949,7 @@ func configureExporter[T any](tb testing.TB,
 		kgo.ClientID(cfg.ClientID),
 	}
 
-	client, err := kafka.NewFranzSyncProducer(tb.Context(), kcfg,
+	client, err := kafka.NewFranzSyncProducer(tb.Context(), host, kcfg,
 		cfg.Producer, 1*time.Second, zap.NewNop(), kgoClientOpts...)
 	require.NoError(tb, err, "failed to create kgo.Client with fake cluster addresses")
 

internal/kafka/franz_client.go

@@ -24,6 +24,7 @@ import (
 	"github.com/twmb/franz-go/pkg/sasl/plain"
 	"github.com/twmb/franz-go/pkg/sasl/scram"
 	"github.com/twmb/franz-go/plugin/kzap"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configcompression"
 	"go.uber.org/zap"
 
@@ -38,7 +39,10 @@ const (
 )
 
 // NewFranzSyncProducer creates a new Kafka client using the franz-go library.
-func NewFranzSyncProducer(ctx context.Context, clientCfg configkafka.ClientConfig,
+func NewFranzSyncProducer(
+	ctx context.Context,
+	host component.Host,
+	clientCfg configkafka.ClientConfig,
 	cfg configkafka.ProducerConfig,
 	timeout time.Duration,
 	logger *zap.Logger,
@@ -50,7 +54,7 @@ func NewFranzSyncProducer(ctx context.Context, clientCfg configkafka.ClientConfi
 	default:
 		codec = codec.WithLevel(int(cfg.CompressionParams.Level))
 	}
-	opts, err := commonOpts(ctx, clientCfg, logger, append(
+	opts, err := commonOpts(ctx, host, clientCfg, logger, append(
 		opts,
 		kgo.ProduceRequestTimeout(timeout),
 		kgo.ProducerBatchCompression(codec),
@@ -86,14 +90,17 @@ func NewFranzSyncProducer(ctx context.Context, clientCfg configkafka.ClientConfi
 }
 
 // NewFranzConsumerGroup creates a new Kafka consumer client using the franz-go library.
-func NewFranzConsumerGroup(ctx context.Context, clientCfg configkafka.ClientConfig,
+func NewFranzConsumerGroup(
+	ctx context.Context,
+	host component.Host,
+	clientCfg configkafka.ClientConfig,
 	consumerCfg configkafka.ConsumerConfig,
 	topics []string,
 	excludeTopics []string,
 	logger *zap.Logger,
 	opts ...kgo.Opt,
 ) (*kgo.Client, error) {
-	opts, err := commonOpts(ctx, clientCfg, logger, append([]kgo.Opt{
+	opts, err := commonOpts(ctx, host, clientCfg, logger, append([]kgo.Opt{
 		kgo.ConsumeTopics(topics...),
 		kgo.ConsumerGroup(consumerCfg.GroupID),
 		kgo.SessionTimeout(consumerCfg.SessionTimeout),
@@ -167,11 +174,12 @@ func NewFranzConsumerGroup(ctx context.Context, clientCfg configkafka.ClientConf
 // NewFranzClient creates a franz-go client using the same commonOpts used for producer/consumer.
 func NewFranzClient(
 	ctx context.Context,
+	host component.Host,
 	clientCfg configkafka.ClientConfig,
 	logger *zap.Logger,
 	opts ...kgo.Opt,
 ) (*kgo.Client, error) {
-	opts, err := commonOpts(ctx, clientCfg, logger, opts...)
+	opts, err := commonOpts(ctx, host, clientCfg, logger, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -181,23 +189,22 @@ func NewFranzClient(
 // NewFranzClusterAdminClient creates a kadm admin client from a freshly created franz client.
 func NewFranzClusterAdminClient(
 	ctx context.Context,
+	host component.Host,
 	clientCfg configkafka.ClientConfig,
 	logger *zap.Logger,
 	opts ...kgo.Opt,
 ) (*kadm.Client, *kgo.Client, error) {
-	cl, err := NewFranzClient(ctx, clientCfg, logger, opts...)
+	cl, err := NewFranzClient(ctx, host, clientCfg, logger, opts...)
 	if err != nil {
 		return nil, nil, err
 	}
 	return kadm.NewClient(cl), cl, nil
 }
 
-// NewFranzAdminFromClient returns a kadm admin bound to an existing kgo client.
-func NewFranzAdminFromClient(cl *kgo.Client) *kadm.Client {
-	return kadm.NewClient(cl)
-}
-
-func commonOpts(ctx context.Context, clientCfg configkafka.ClientConfig,
+func commonOpts(
+	ctx context.Context,
+	_ component.Host,
+	clientCfg configkafka.ClientConfig,
 	logger *zap.Logger,
 	opts ...kgo.Opt,
 ) ([]kgo.Opt, error) {

internal/kafka/franz_client_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/twmb/franz-go/pkg/kfake"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/kmsg"
+	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.uber.org/zap"
@@ -45,7 +46,8 @@ func TestNewFranzSyncProducer_SASL(t *testing.T) {
 			Version:   1, // kfake only supports version 1
 		}
 		tl := zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel))
-		client, err := NewFranzSyncProducer(t.Context(), clientConfig,
+		client, err := NewFranzSyncProducer(
+			t.Context(), componenttest.NewNopHost(), clientConfig,
 			configkafka.NewDefaultProducerConfig(), time.Second, tl,
 		)
 		if err != nil {
@@ -124,7 +126,8 @@ func TestNewFranzSyncProducer_TLS(t *testing.T) {
 		observedLogs.TakeAll()       // drop existing logs
 		clientConfig := clientConfig // copy
 		clientConfig.TLS = cfg
-		client, err := NewFranzSyncProducer(t.Context(), clientConfig,
+		client, err := NewFranzSyncProducer(
+			t.Context(), componenttest.NewNopHost(), clientConfig,
 			configkafka.NewDefaultProducerConfig(), time.Second, logger,
 		)
 		if err != nil {
@@ -183,7 +186,10 @@ func TestNewFranzSyncProducerCompression(t *testing.T) {
 			prodCfg.Compression = compressionAlgo
 
 			tl := zaptest.NewLogger(t, zaptest.Level(zap.InfoLevel))
-			client, err := NewFranzSyncProducer(t.Context(), clientConfig, prodCfg, time.Second, tl)
+			client, err := NewFranzSyncProducer(
+				t.Context(), componenttest.NewNopHost(), clientConfig,
+				prodCfg, time.Second, tl,
+			)
 			require.NoError(t, err)
 			defer client.Close()
 
@@ -250,7 +256,10 @@ func TestNewFranzSyncProducerRequiredAcks(t *testing.T) {
 			prodCfg.RequiredAcks = ack
 
 			tl := zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel))
-			client, err := NewFranzSyncProducer(t.Context(), clientConfig, prodCfg, time.Second, tl)
+			client, err := NewFranzSyncProducer(
+				t.Context(), componenttest.NewNopHost(), clientConfig,
+				prodCfg, time.Second, tl,
+			)
 			require.NoError(t, err)
 			defer client.Close()
 
@@ -425,7 +434,8 @@ func mustNewFranzConsumerGroup(t *testing.T,
 	// up and avoid waiting for too long.
 	minAge := 10 * time.Millisecond
 	opts = append(opts, kgo.MetadataMinAge(minAge), kgo.MetadataMaxAge(minAge*2))
-	client, err := NewFranzConsumerGroup(t.Context(), clientConfig, consumerConfig,
+	client, err := NewFranzConsumerGroup(
+		t.Context(), componenttest.NewNopHost(), clientConfig, consumerConfig,
 		topics, nil, zaptest.NewLogger(t, zaptest.Level(zap.InfoLevel)), opts...,
 	)
 	require.NoError(t, err)
@@ -447,7 +457,8 @@ func TestFranzClient_MetadataRefreshInterval(t *testing.T) {
 			name: "producer",
 			setupClient: func(t *testing.T, clientConfig configkafka.ClientConfig, _ string, metadataMinAge time.Duration) {
 				tl := zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel))
-				client, err := NewFranzSyncProducer(t.Context(), clientConfig,
+				client, err := NewFranzSyncProducer(
+					t.Context(), componenttest.NewNopHost(), clientConfig,
 					configkafka.NewDefaultProducerConfig(), time.Second, tl,
 					kgo.MetadataMinAge(metadataMinAge),
 				)
@@ -534,7 +545,7 @@ func TestFranzClient_ProtocolVersion(t *testing.T) {
 			})
 			t.Run("producer", func(t *testing.T) {
 				client, err := NewFranzSyncProducer(
-					t.Context(), clientConfig,
+					t.Context(), componenttest.NewNopHost(), clientConfig,
 					configkafka.NewDefaultProducerConfig(), time.Second, zap.NewNop(),
 				)
 				require.NoError(t, err)
@@ -551,12 +562,12 @@ func TestNewFranzClient_And_Admin(t *testing.T) {
 	tl := zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel))
 
 	// Plain client
-	cl, err := NewFranzClient(t.Context(), clientCfg, tl)
+	cl, err := NewFranzClient(t.Context(), componenttest.NewNopHost(), clientCfg, tl)
 	require.NoError(t, err)
 	t.Cleanup(cl.Close)
 
 	// Admin from fresh client
-	ad, cl2, err := NewFranzClusterAdminClient(t.Context(), clientCfg, tl)
+	ad, cl2, err := NewFranzClusterAdminClient(t.Context(), componenttest.NewNopHost(), clientCfg, tl)
 	require.NoError(t, err)
 	t.Cleanup(func() { ad.Close(); cl2.Close() })
 

internal/kafka/go.mod

@@ -11,14 +11,14 @@ require (
 	github.com/twmb/franz-go/pkg/kfake v0.0.0-20251021233722-4ca18825d8c0
 	github.com/twmb/franz-go/pkg/sasl/kerberos v1.1.0
 	github.com/xdg-go/scram v1.2.0
+	go.opentelemetry.io/collector/component v1.49.1-0.20260119125341-0550b08ddcc5
+	go.opentelemetry.io/collector/component/componenttest v0.143.1-0.20260119125341-0550b08ddcc5
 	go.opentelemetry.io/collector/config/configcompression v1.49.1-0.20260119125341-0550b08ddcc5
 	go.opentelemetry.io/collector/config/configopaque v1.49.1-0.20260119125341-0550b08ddcc5
 	go.opentelemetry.io/collector/config/configtls v1.49.1-0.20260119125341-0550b08ddcc5
 	go.uber.org/goleak v1.3.0
 )
 
-require go.opentelemetry.io/collector/confmap/xconfmap v0.143.1-0.20260119125341-0550b08ddcc5 // indirect
-
 require (
 	github.com/aws/aws-sdk-go-v2 v1.36.4 // indirect
 	github.com/aws/aws-sdk-go-v2/config v1.29.16 // indirect
@@ -33,46 +33,60 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.2 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.33.21 // indirect
 	github.com/aws/smithy-go v1.22.2 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/eapache/go-resiliency v1.7.0 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/foxboron/go-tpm-keyfiles v0.0.0-20250903184740-5d135037bd4d // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-viper/mapstructure/v2 v2.5.0 // indirect
 	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/go-tpm v0.9.8 // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
 	github.com/hashicorp/go-version v1.8.0 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
 	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.4
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.18.2 // indirect
 	github.com/knadh/koanf/maps v0.1.2 // indirect
 	github.com/knadh/koanf/providers/confmap v1.0.0 // indirect
 	github.com/knadh/koanf/v2 v2.3.0 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/rogpeppe/go-internal v1.13.1 // indirect
 	github.com/twmb/franz-go v1.20.6
 	github.com/twmb/franz-go/pkg/kadm v1.17.1
 	github.com/twmb/franz-go/pkg/kmsg v1.12.0
 	github.com/twmb/franz-go/plugin/kzap v1.1.2
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
+	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/collector/confmap v1.49.1-0.20260119125341-0550b08ddcc5 // indirect
+	go.opentelemetry.io/collector/confmap/xconfmap v0.143.1-0.20260119125341-0550b08ddcc5 // indirect
 	go.opentelemetry.io/collector/featuregate v1.49.1-0.20260119125341-0550b08ddcc5 // indirect
+	go.opentelemetry.io/collector/pdata v1.49.1-0.20260119125341-0550b08ddcc5 // indirect
+	go.opentelemetry.io/otel v1.39.0 // indirect
+	go.opentelemetry.io/otel/metric v1.39.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.39.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.39.0 // indirect
+	go.opentelemetry.io/otel/trace v1.39.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.1
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/crypto v0.45.0 // indirect
 	golang.org/x/net v0.47.0 // indirect
-	golang.org/x/sys v0.38.0 // indirect
+	golang.org/x/sys v0.39.0 // indirect
 	golang.org/x/text v0.31.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )