[receiver/kafkareceiver] Add encoding extensions support (#33888)

**Description:** Add support for encoding extensions in the
kafkareceiver
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.-->
To be able to use encoding extensions this PR adds extension support and
proposes to rename the existing `encoding` configuration property to
`format` and reusing the `encoding` property for configuring encoding
extensions. Reason is to be consistent with other receivers/exporters
like the `fileexporter` that already support extensions.

**Link to tracking Issue:** n/a

**Testing:** Tested with the existing avro_log_encoding extension as
well with receivers internal json encoding.

**Documentation:**: Updated README.md within the receiver describing the
use of encoding extensions.

.chloggen/kafkareceiver-encoding-extensions.yaml

@@ -0,0 +1,28 @@
+# 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: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for encoding extensions in the Kafka receiver.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33888]
+
+# (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 support for encoding extensions in the Kafka receiver. Loading extensions takes precedence over the internally supported encodings.
+
+# 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]

receiver/kafkareceiver/README.md

@@ -29,7 +29,7 @@ The following settings can be optionally configured:
 - `resolve_canonical_bootstrap_servers_only` (default = false): Whether to resolve then reverse-lookup broker IPs during startup
 - `topic` (default = otlp_spans for traces, otlp_metrics for metrics, otlp_logs for logs): The name of the kafka topic to read from.
   Only one telemetry type may be used for a given topic.
-- `encoding` (default = otlp_proto): The encoding of the payload received from kafka. Available encodings:
+- `encoding` (default = otlp_proto): The encoding of the payload received from kafka. Supports encoding extensions. Tries to load an encoding extension and falls back to internal encodings if no extension was loaded. Available internal encodings:
   - `otlp_proto`: the payload is deserialized to `ExportTraceServiceRequest`, `ExportLogsServiceRequest` or `ExportMetricsServiceRequest` respectively.
   - `jaeger_proto`: the payload is deserialized to a single Jaeger proto `Span`.
   - `jaeger_json`: the payload is deserialized to a single Jaeger JSON Span using `jsonpb`.

receiver/kafkareceiver/factory.go

@@ -54,40 +54,9 @@ var errUnrecognizedEncoding = fmt.Errorf("unrecognized encoding")
 // FactoryOption applies changes to kafkaExporterFactory.
 type FactoryOption func(factory *kafkaReceiverFactory)
 
-// withTracesUnmarshalers adds Unmarshalers.
-func withTracesUnmarshalers(tracesUnmarshalers ...TracesUnmarshaler) FactoryOption {
-	return func(factory *kafkaReceiverFactory) {
-		for _, unmarshaler := range tracesUnmarshalers {
-			factory.tracesUnmarshalers[unmarshaler.Encoding()] = unmarshaler
-		}
-	}
-}
-
-// withMetricsUnmarshalers adds MetricsUnmarshalers.
-func withMetricsUnmarshalers(metricsUnmarshalers ...MetricsUnmarshaler) FactoryOption {
-	return func(factory *kafkaReceiverFactory) {
-		for _, unmarshaler := range metricsUnmarshalers {
-			factory.metricsUnmarshalers[unmarshaler.Encoding()] = unmarshaler
-		}
-	}
-}
-
-// withLogsUnmarshalers adds LogsUnmarshalers.
-func withLogsUnmarshalers(logsUnmarshalers ...LogsUnmarshaler) FactoryOption {
-	return func(factory *kafkaReceiverFactory) {
-		for _, unmarshaler := range logsUnmarshalers {
-			factory.logsUnmarshalers[unmarshaler.Encoding()] = unmarshaler
-		}
-	}
-}
-
 // NewFactory creates Kafka receiver factory.
 func NewFactory(options ...FactoryOption) receiver.Factory {
-	f := &kafkaReceiverFactory{
-		tracesUnmarshalers:  map[string]TracesUnmarshaler{},
-		metricsUnmarshalers: map[string]MetricsUnmarshaler{},
-		logsUnmarshalers:    map[string]LogsUnmarshaler{},
-	}
+	f := &kafkaReceiverFactory{}
 	for _, o := range options {
 		o(f)
 	}
@@ -133,32 +102,20 @@ func createDefaultConfig() component.Config {
 	}
 }
 
-type kafkaReceiverFactory struct {
-	tracesUnmarshalers  map[string]TracesUnmarshaler
-	metricsUnmarshalers map[string]MetricsUnmarshaler
-	logsUnmarshalers    map[string]LogsUnmarshaler
-}
+type kafkaReceiverFactory struct{}
 
 func (f *kafkaReceiverFactory) createTracesReceiver(
 	_ context.Context,
 	set receiver.Settings,
 	cfg component.Config,
 	nextConsumer consumer.Traces,
 ) (receiver.Traces, error) {
-	for encoding, unmarshal := range defaultTracesUnmarshalers() {
-		f.tracesUnmarshalers[encoding] = unmarshal
-	}
-
 	oCfg := *(cfg.(*Config))
 	if oCfg.Topic == "" {
 		oCfg.Topic = defaultTracesTopic
 	}
-	unmarshaler := f.tracesUnmarshalers[oCfg.Encoding]
-	if unmarshaler == nil {
-		return nil, errUnrecognizedEncoding
-	}
 
-	r, err := newTracesReceiver(oCfg, set, unmarshaler, nextConsumer)
+	r, err := newTracesReceiver(oCfg, set, nextConsumer)
 	if err != nil {
 		return nil, err
 	}
@@ -171,20 +128,12 @@ func (f *kafkaReceiverFactory) createMetricsReceiver(
 	cfg component.Config,
 	nextConsumer consumer.Metrics,
 ) (receiver.Metrics, error) {
-	for encoding, unmarshal := range defaultMetricsUnmarshalers() {
-		f.metricsUnmarshalers[encoding] = unmarshal
-	}
-
 	oCfg := *(cfg.(*Config))
 	if oCfg.Topic == "" {
 		oCfg.Topic = defaultMetricsTopic
 	}
-	unmarshaler := f.metricsUnmarshalers[oCfg.Encoding]
-	if unmarshaler == nil {
-		return nil, errUnrecognizedEncoding
-	}
 
-	r, err := newMetricsReceiver(oCfg, set, unmarshaler, nextConsumer)
+	r, err := newMetricsReceiver(oCfg, set, nextConsumer)
 	if err != nil {
 		return nil, err
 	}
@@ -197,20 +146,12 @@ func (f *kafkaReceiverFactory) createLogsReceiver(
 	cfg component.Config,
 	nextConsumer consumer.Logs,
 ) (receiver.Logs, error) {
-	for encoding, unmarshaler := range defaultLogsUnmarshalers(set.BuildInfo.Version, set.Logger) {
-		f.logsUnmarshalers[encoding] = unmarshaler
-	}
-
 	oCfg := *(cfg.(*Config))
 	if oCfg.Topic == "" {
 		oCfg.Topic = defaultLogsTopic
 	}
-	unmarshaler, err := getLogsUnmarshaler(oCfg.Encoding, f.logsUnmarshalers)
-	if err != nil {
-		return nil, err
-	}
 
-	r, err := newLogsReceiver(oCfg, set, unmarshaler, nextConsumer)
+	r, err := newLogsReceiver(oCfg, set, nextConsumer)
 	if err != nil {
 		return nil, err
 	}

receiver/kafkareceiver/factory_test.go

@@ -37,7 +37,7 @@ func TestCreateTracesReceiver(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
 	cfg.Brokers = []string{"invalid:9092"}
 	cfg.ProtocolVersion = "2.0.0"
-	f := kafkaReceiverFactory{tracesUnmarshalers: defaultTracesUnmarshalers()}
+	f := kafkaReceiverFactory{}
 	r, err := f.createTracesReceiver(context.Background(), receivertest.NewNopSettings(), cfg, nil)
 	require.NoError(t, err)
 	// no available broker
@@ -46,7 +46,7 @@ func TestCreateTracesReceiver(t *testing.T) {
 
 func TestWithTracesUnmarshalers(t *testing.T) {
 	unmarshaler := &customTracesUnmarshaler{}
-	f := NewFactory(withTracesUnmarshalers(unmarshaler))
+	f := NewFactory()
 	cfg := createDefaultConfig().(*Config)
 	// disable contacting broker
 	cfg.Metadata.Full = false
@@ -76,7 +76,7 @@ func TestCreateMetricsReceiver(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
 	cfg.Brokers = []string{"invalid:9092"}
 	cfg.ProtocolVersion = "2.0.0"
-	f := kafkaReceiverFactory{metricsUnmarshalers: defaultMetricsUnmarshalers()}
+	f := kafkaReceiverFactory{}
 	r, err := f.createMetricsReceiver(context.Background(), receivertest.NewNopSettings(), cfg, nil)
 	require.NoError(t, err)
 	// no available broker
@@ -85,7 +85,7 @@ func TestCreateMetricsReceiver(t *testing.T) {
 
 func TestWithMetricsUnmarshalers(t *testing.T) {
 	unmarshaler := &customMetricsUnmarshaler{}
-	f := NewFactory(withMetricsUnmarshalers(unmarshaler))
+	f := NewFactory()
 	cfg := createDefaultConfig().(*Config)
 	// disable contacting broker
 	cfg.Metadata.Full = false
@@ -115,7 +115,7 @@ func TestCreateLogsReceiver(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
 	cfg.Brokers = []string{"invalid:9092"}
 	cfg.ProtocolVersion = "2.0.0"
-	f := kafkaReceiverFactory{logsUnmarshalers: defaultLogsUnmarshalers("Test Version", zap.NewNop())}
+	f := kafkaReceiverFactory{}
 	r, err := f.createLogsReceiver(context.Background(), receivertest.NewNopSettings(), cfg, nil)
 	require.NoError(t, err)
 	// no available broker
@@ -146,7 +146,7 @@ func TestGetLogsUnmarshaler_encoding_text_error(t *testing.T) {
 
 func TestWithLogsUnmarshalers(t *testing.T) {
 	unmarshaler := &customLogsUnmarshaler{}
-	f := NewFactory(withLogsUnmarshalers(unmarshaler))
+	f := NewFactory()
 	cfg := createDefaultConfig().(*Config)
 	// disable contacting broker
 	cfg.Metadata.Full = false

receiver/kafkareceiver/kafka_receiver.go

@@ -14,6 +14,9 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componentstatus"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
 	"go.opentelemetry.io/otel/attribute"
@@ -100,11 +103,7 @@ var _ receiver.Traces = (*kafkaTracesConsumer)(nil)
 var _ receiver.Metrics = (*kafkaMetricsConsumer)(nil)
 var _ receiver.Logs = (*kafkaLogsConsumer)(nil)
 
-func newTracesReceiver(config Config, set receiver.Settings, unmarshaler TracesUnmarshaler, nextConsumer consumer.Traces) (*kafkaTracesConsumer, error) {
-	if unmarshaler == nil {
-		return nil, errUnrecognizedEncoding
-	}
-
+func newTracesReceiver(config Config, set receiver.Settings, nextConsumer consumer.Traces) (*kafkaTracesConsumer, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(set.TelemetrySettings)
 	if err != nil {
 		return nil, err
@@ -114,7 +113,6 @@ func newTracesReceiver(config Config, set receiver.Settings, unmarshaler TracesU
 		config:            config,
 		topics:            []string{config.Topic},
 		nextConsumer:      nextConsumer,
-		unmarshaler:       unmarshaler,
 		settings:          set,
 		autocommitEnabled: config.AutoCommit.Enable,
 		messageMarking:    config.MessageMarking,
@@ -170,6 +168,22 @@ func (c *kafkaTracesConsumer) Start(_ context.Context, host component.Host) erro
 	if err != nil {
 		return err
 	}
+	// extensions take precedence over internal encodings
+	if unmarshaler, errExt := loadEncodingExtension[ptrace.Unmarshaler](
+		host,
+		c.config.Encoding,
+	); errExt == nil {
+		c.unmarshaler = &tracesEncodingUnmarshaler{
+			unmarshaler: *unmarshaler,
+			encoding:    c.config.Encoding,
+		}
+	}
+	if unmarshaler, ok := defaultTracesUnmarshalers()[c.config.Encoding]; c.unmarshaler == nil && ok {
+		c.unmarshaler = unmarshaler
+	}
+	if c.unmarshaler == nil {
+		return errUnrecognizedEncoding
+	}
 	// consumerGroup may be set in tests to inject fake implementation.
 	if c.consumerGroup == nil {
 		if c.consumerGroup, err = createKafkaClient(c.config); err != nil {
@@ -229,11 +243,7 @@ func (c *kafkaTracesConsumer) Shutdown(context.Context) error {
 	return c.consumerGroup.Close()
 }
 
-func newMetricsReceiver(config Config, set receiver.Settings, unmarshaler MetricsUnmarshaler, nextConsumer consumer.Metrics) (*kafkaMetricsConsumer, error) {
-	if unmarshaler == nil {
-		return nil, errUnrecognizedEncoding
-	}
-
+func newMetricsReceiver(config Config, set receiver.Settings, nextConsumer consumer.Metrics) (*kafkaMetricsConsumer, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(set.TelemetrySettings)
 	if err != nil {
 		return nil, err
@@ -243,7 +253,6 @@ func newMetricsReceiver(config Config, set receiver.Settings, unmarshaler Metric
 		config:            config,
 		topics:            []string{config.Topic},
 		nextConsumer:      nextConsumer,
-		unmarshaler:       unmarshaler,
 		settings:          set,
 		autocommitEnabled: config.AutoCommit.Enable,
 		messageMarking:    config.MessageMarking,
@@ -267,6 +276,22 @@ func (c *kafkaMetricsConsumer) Start(_ context.Context, host component.Host) err
 	if err != nil {
 		return err
 	}
+	// extensions take precedence over internal encodings
+	if unmarshaler, errExt := loadEncodingExtension[pmetric.Unmarshaler](
+		host,
+		c.config.Encoding,
+	); errExt == nil {
+		c.unmarshaler = &metricsEncodingUnmarshaler{
+			unmarshaler: *unmarshaler,
+			encoding:    c.config.Encoding,
+		}
+	}
+	if unmarshaler, ok := defaultMetricsUnmarshalers()[c.config.Encoding]; c.unmarshaler == nil && ok {
+		c.unmarshaler = unmarshaler
+	}
+	if c.unmarshaler == nil {
+		return errUnrecognizedEncoding
+	}
 	// consumerGroup may be set in tests to inject fake implementation.
 	if c.consumerGroup == nil {
 		if c.consumerGroup, err = createKafkaClient(c.config); err != nil {
@@ -326,11 +351,7 @@ func (c *kafkaMetricsConsumer) Shutdown(context.Context) error {
 	return c.consumerGroup.Close()
 }
 
-func newLogsReceiver(config Config, set receiver.Settings, unmarshaler LogsUnmarshaler, nextConsumer consumer.Logs) (*kafkaLogsConsumer, error) {
-	if unmarshaler == nil {
-		return nil, errUnrecognizedEncoding
-	}
-
+func newLogsReceiver(config Config, set receiver.Settings, nextConsumer consumer.Logs) (*kafkaLogsConsumer, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(set.TelemetrySettings)
 	if err != nil {
 		return nil, err
@@ -340,7 +361,6 @@ func newLogsReceiver(config Config, set receiver.Settings, unmarshaler LogsUnmar
 		config:            config,
 		topics:            []string{config.Topic},
 		nextConsumer:      nextConsumer,
-		unmarshaler:       unmarshaler,
 		settings:          set,
 		autocommitEnabled: config.AutoCommit.Enable,
 		messageMarking:    config.MessageMarking,
@@ -364,6 +384,25 @@ func (c *kafkaLogsConsumer) Start(_ context.Context, host component.Host) error
 	if err != nil {
 		return err
 	}
+	// extensions take precedence over internal encodings
+	if unmarshaler, errExt := loadEncodingExtension[plog.Unmarshaler](
+		host,
+		c.config.Encoding,
+	); errExt == nil {
+		c.unmarshaler = &logsEncodingUnmarshaler{
+			unmarshaler: *unmarshaler,
+			encoding:    c.config.Encoding,
+		}
+	}
+	if unmarshaler, errInt := getLogsUnmarshaler(
+		c.config.Encoding,
+		defaultLogsUnmarshalers(c.settings.BuildInfo.Version, c.settings.Logger),
+	); c.unmarshaler == nil && errInt == nil {
+		c.unmarshaler = unmarshaler
+	}
+	if c.unmarshaler == nil {
+		return errUnrecognizedEncoding
+	}
 	// consumerGroup may be set in tests to inject fake implementation.
 	if c.consumerGroup == nil {
 		if c.consumerGroup, err = createKafkaClient(c.config); err != nil {
@@ -720,3 +759,30 @@ func toSaramaInitialOffset(initialOffset string) (int64, error) {
 		return 0, errInvalidInitialOffset
 	}
 }
+
+// loadEncodingExtension tries to load an available extension for the given encoding.
+func loadEncodingExtension[T any](host component.Host, encoding string) (*T, error) {
+	extensionID, err := encodingToComponentID(encoding)
+	if err != nil {
+		return nil, err
+	}
+	encodingExtension, ok := host.GetExtensions()[*extensionID]
+	if !ok {
+		return nil, fmt.Errorf("unknown encoding extension %q", encoding)
+	}
+	unmarshaler, ok := encodingExtension.(T)
+	if !ok {
+		return nil, fmt.Errorf("extension %q is not an unmarshaler", encoding)
+	}
+	return &unmarshaler, nil
+}
+
+// encodingToComponentID converts an encoding string to a component ID using the given encoding as type.
+func encodingToComponentID(encoding string) (*component.ID, error) {
+	componentType, err := component.NewType(encoding)
+	if err != nil {
+		return nil, fmt.Errorf("invalid component type: %w", err)
+	}
+	id := component.NewID(componentType)
+	return &id, nil
+}