exporter/kafkaexporter: tidy up; support partitioning in all encodings

exporter/kafkaexporter/config.go

@@ -31,11 +31,18 @@ type Config struct {
 	Encoding string `mapstructure:"encoding"`
 
 	// PartitionTracesByID sets the message key of outgoing trace messages to the trace ID.
-	// Please note: does not have any effect on Jaeger encoding exporters since Jaeger exporters include
-	// trace ID as the message key by default.
+	//
+	// NOTE: this does not have any effect for Jaeger encodings. Jaeger encodings always use
+	// use the trace ID for the message key.
 	PartitionTracesByID bool `mapstructure:"partition_traces_by_id"`
 
+	// PartitionMetricsByResourceAttributes controls the partitioning of metrics messages by
+	// resource. If this is true, then the message key will be set to a hash of the resource's
+	// identifying attributes.
 	PartitionMetricsByResourceAttributes bool `mapstructure:"partition_metrics_by_resource_attributes"`
 
+	// PartitionLogsByResourceAttributes controls the partitioning of logs messages by resource.
+	// If this is true, then the message key will be set to a hash of the resource's identifying
+	// attributes.
 	PartitionLogsByResourceAttributes bool `mapstructure:"partition_logs_by_resource_attributes"`
 }

exporter/kafkaexporter/factory.go

@@ -70,15 +70,16 @@ func createTracesExporter(
 		ctx,
 		set,
 		&oCfg,
-		exp.tracesPusher,
+		exp.exportData,
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		// Disable exporterhelper Timeout, because we cannot pass a Context to the Producer,
 		// and will rely on the sarama Producer Timeout logic.
 		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 		exporterhelper.WithRetry(oCfg.BackOffConfig),
 		exporterhelper.WithQueue(oCfg.QueueSettings),
-		exporterhelper.WithStart(exp.start),
-		exporterhelper.WithShutdown(exp.Close))
+		exporterhelper.WithStart(exp.Start),
+		exporterhelper.WithShutdown(exp.Close),
+	)
 }
 
 func createMetricsExporter(
@@ -98,15 +99,16 @@ func createMetricsExporter(
 		ctx,
 		set,
 		&oCfg,
-		exp.metricsDataPusher,
+		exp.exportData,
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		// Disable exporterhelper Timeout, because we cannot pass a Context to the Producer,
 		// and will rely on the sarama Producer Timeout logic.
 		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 		exporterhelper.WithRetry(oCfg.BackOffConfig),
 		exporterhelper.WithQueue(oCfg.QueueSettings),
-		exporterhelper.WithStart(exp.start),
-		exporterhelper.WithShutdown(exp.Close))
+		exporterhelper.WithStart(exp.Start),
+		exporterhelper.WithShutdown(exp.Close),
+	)
 }
 
 func createLogsExporter(
@@ -126,13 +128,14 @@ func createLogsExporter(
 		ctx,
 		set,
 		&oCfg,
-		exp.logsDataPusher,
+		exp.exportData,
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		// Disable exporterhelper Timeout, because we cannot pass a Context to the Producer,
 		// and will rely on the sarama Producer Timeout logic.
 		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 		exporterhelper.WithRetry(oCfg.BackOffConfig),
 		exporterhelper.WithQueue(oCfg.QueueSettings),
-		exporterhelper.WithStart(exp.start),
-		exporterhelper.WithShutdown(exp.Close))
+		exporterhelper.WithStart(exp.Start),
+		exporterhelper.WithShutdown(exp.Close),
+	)
 }

exporter/kafkaexporter/factory_test.go

@@ -38,10 +38,9 @@ func TestCreateMetricExporter(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		name       string
-		conf       *Config
-		marshalers []MetricsMarshaler
-		err        *net.DNSError
+		name string
+		conf *Config
+		err  *net.DNSError
 	}{
 		{
 			name: "valid config (no validating broker)",
@@ -69,8 +68,7 @@ func TestCreateMetricExporter(t *testing.T) {
 				conf.Metadata.Full = false
 				conf.Encoding = defaultEncoding
 			}),
-			marshalers: nil,
-			err:        nil,
+			err: nil,
 		},
 	}
 
@@ -102,10 +100,9 @@ func TestCreateLogExporter(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		name       string
-		conf       *Config
-		marshalers []LogsMarshaler
-		err        *net.DNSError
+		name string
+		conf *Config
+		err  *net.DNSError
 	}{
 		{
 			name: "valid config (no validating broker)",
@@ -133,8 +130,7 @@ func TestCreateLogExporter(t *testing.T) {
 				conf.Metadata.Full = false
 				conf.Encoding = defaultEncoding
 			}),
-			marshalers: nil,
-			err:        nil,
+			err: nil,
 		},
 	}
 
@@ -166,10 +162,9 @@ func TestCreateTraceExporter(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		name       string
-		conf       *Config
-		marshalers []TracesMarshaler
-		err        *net.DNSError
+		name string
+		conf *Config
+		err  *net.DNSError
 	}{
 		{
 			name: "valid config (no validating brokers)",
@@ -178,17 +173,15 @@ func TestCreateTraceExporter(t *testing.T) {
 				conf.Brokers = []string{"invalid:9092"}
 				conf.ProtocolVersion = "2.0.0"
 			}),
-			marshalers: nil,
-			err:        nil,
+			err: nil,
 		},
 		{
 			name: "invalid config (validating brokers)",
 			conf: applyConfigOption(func(conf *Config) {
 				conf.Brokers = []string{"invalid:9092"}
 				conf.ProtocolVersion = "2.0.0"
 			}),
-			marshalers: nil,
-			err:        &net.DNSError{},
+			err: &net.DNSError{},
 		},
 		{
 			name: "default_encoding",
@@ -197,8 +190,7 @@ func TestCreateTraceExporter(t *testing.T) {
 				conf.Metadata.Full = false
 				conf.Encoding = defaultEncoding
 			}),
-			marshalers: nil,
-			err:        nil,
+			err: nil,
 		},
 	}
 

exporter/kafkaexporter/go.mod

@@ -10,10 +10,10 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka v0.122.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.122.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic v0.122.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.122.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.122.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.122.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin v0.122.0
-	github.com/openzipkin/zipkin-go v0.4.3
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v1.28.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/component/componenttest v0.122.2-0.20250319144947-41a9ea7f7402
@@ -26,7 +26,6 @@ require (
 	go.opentelemetry.io/collector/exporter/exportertest v0.122.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/pdata v1.28.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/pdata/testdata v0.122.2-0.20250319144947-41a9ea7f7402
-	go.opentelemetry.io/collector/semconv v0.122.2-0.20250319144947-41a9ea7f7402
 	go.uber.org/goleak v1.3.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
@@ -79,6 +78,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/core/xidutils v0.122.0 // indirect
+	github.com/openzipkin/zipkin-go v0.4.3 // indirect
 	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
@@ -99,6 +99,7 @@ require (
 	go.opentelemetry.io/collector/receiver v1.28.2-0.20250319144947-41a9ea7f7402 // indirect
 	go.opentelemetry.io/collector/receiver/receivertest v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
 	go.opentelemetry.io/collector/receiver/xreceiver v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
+	go.opentelemetry.io/collector/semconv v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
 	go.opentelemetry.io/otel v1.35.0 // indirect
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.35.0 // indirect

exporter/kafkaexporter/internal/marshaler/jaeger_marshaler.go

@@ -0,0 +1,66 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package marshaler // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/marshaler"
+
+import (
+	"bytes"
+
+	"github.com/gogo/protobuf/jsonpb"
+	jaegerproto "github.com/jaegertracing/jaeger-idl/model/v1"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/multierr"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger"
+)
+
+var (
+	_ TracesMarshaler = JaegerProtoSpanMarshaler{}
+	_ TracesMarshaler = JaegerJSONSpanMarshaler{}
+)
+
+type JaegerProtoSpanMarshaler struct{}
+
+type JaegerJSONSpanMarshaler struct{}
+
+func (JaegerProtoSpanMarshaler) MarshalTraces(traces ptrace.Traces) ([]Message, error) {
+	return marshalJaeger(traces, marshalJaegerSpanProto)
+}
+
+func (JaegerJSONSpanMarshaler) MarshalTraces(traces ptrace.Traces) ([]Message, error) {
+	return marshalJaeger(traces, marshalJaegerSpanJSON)
+}
+
+func marshalJaeger(traces ptrace.Traces, marshal marshalJaegerSpanFunc) ([]Message, error) {
+	batches := jaeger.ProtoFromTraces(traces)
+	var messages []Message
+
+	var errs error
+	for _, batch := range batches {
+		for _, span := range batch.Spans {
+			span.Process = batch.Process
+			bts, err := marshal(span)
+			// continue to process spans that can be serialized
+			if err != nil {
+				errs = multierr.Append(errs, err)
+				continue
+			}
+			key := []byte(span.TraceID.String())
+			messages = append(messages, Message{Key: key, Value: bts})
+		}
+	}
+	return messages, errs
+}
+
+type marshalJaegerSpanFunc func(*jaegerproto.Span) ([]byte, error)
+
+func marshalJaegerSpanProto(span *jaegerproto.Span) ([]byte, error) {
+	return span.Marshal()
+}
+
+func marshalJaegerSpanJSON(span *jaegerproto.Span) ([]byte, error) {
+	var m jsonpb.Marshaler
+	out := new(bytes.Buffer)
+	err := m.Marshal(out, span)
+	return out.Bytes(), err
+}

exporter/kafkaexporter/jaeger_marshaler_test.go → exporter/kafkaexporter/internal/marshaler/jaeger_marshaler_test.go

@@ -1,13 +1,12 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-package kafkaexporter
+package marshaler
 
 import (
 	"bytes"
 	"testing"
 
-	"github.com/IBM/sarama"
 	"github.com/gogo/protobuf/jsonpb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -38,33 +37,24 @@ func TestJaegerMarshaler(t *testing.T) {
 	require.NoError(t, jsonMarshaler.Marshal(jsonByteBuffer, batches[0].Spans[0]))
 
 	tests := []struct {
-		unmarshaler TracesMarshaler
-		encoding    string
-		messages    []*sarama.ProducerMessage
+		marshaler TracesMarshaler
+		encoding  string
+		messages  []Message
 	}{
 		{
-			unmarshaler: jaegerMarshaler{
-				marshaler: jaegerProtoSpanMarshaler{},
-			},
-			encoding: "jaeger_proto",
-			messages: []*sarama.ProducerMessage{{Topic: "topic", Value: sarama.ByteEncoder(jaegerProtoBytes), Key: sarama.ByteEncoder(messageKey)}},
+			marshaler: JaegerProtoSpanMarshaler{},
+			messages:  []Message{{Value: jaegerProtoBytes, Key: messageKey}},
 		},
 		{
-			unmarshaler: jaegerMarshaler{
-				marshaler: jaegerJSONSpanMarshaler{
-					pbMarshaler: &jsonpb.Marshaler{},
-				},
-			},
-			encoding: "jaeger_json",
-			messages: []*sarama.ProducerMessage{{Topic: "topic", Value: sarama.ByteEncoder(jsonByteBuffer.Bytes()), Key: sarama.ByteEncoder(messageKey)}},
+			marshaler: JaegerJSONSpanMarshaler{},
+			messages:  []Message{{Value: jsonByteBuffer.Bytes(), Key: messageKey}},
 		},
 	}
 	for _, test := range tests {
 		t.Run(test.encoding, func(t *testing.T) {
-			messages, err := test.unmarshaler.Marshal(td, "topic")
+			messages, err := test.marshaler.MarshalTraces(td)
 			require.NoError(t, err)
 			assert.Equal(t, test.messages, messages)
-			assert.Equal(t, test.encoding, test.unmarshaler.Encoding())
 		})
 	}
 }

exporter/kafkaexporter/internal/marshaler/marshaler.go

@@ -0,0 +1,40 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package marshaler // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/marshaler"
+
+import (
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+// Message represents a Kafka message.
+//
+// Note that the topic and message headers are set by the Kafka exporter
+// code, and not by the marshaler.
+type Message struct {
+	// Key is an optional message key.
+	//
+	// Marshalers may set this, but it is generally expected that the
+	// Kafka producer will set this based partition_* configuration.
+	Key []byte
+
+	// Value is the message payload.
+	Value []byte
+}
+
+// TracesMarshaler marshals a ptrace.Traces into one or more Messages.
+type TracesMarshaler interface {
+	MarshalTraces(traces ptrace.Traces) ([]Message, error)
+}
+
+// MetricsMarshaler marshals a pmetric.Metrics into one or more Messages.
+type MetricsMarshaler interface {
+	MarshalMetrics(metrics pmetric.Metrics) ([]Message, error)
+}
+
+// LogsMarshaler marshals a plog.Logs into one or more Messages.
+type LogsMarshaler interface {
+	MarshalLogs(logs plog.Logs) ([]Message, error)
+}