kafkareceiver: add signal-specific topic/encoding (#38985)

#### Description

Add signal-specific configuration for topic and encoding.

The topics are already signal-specific by default, it just hasn't been
possible to explicitly configure a different topic for each signal. Thus
if you set the `topic: foo`, it would be used for all signals, which is
never going to work with the receiver.

Similarly, while the default encoding is the same for all signals (i.e.
otlp_proto), some encodings are available only for certain signals, e.g.
azure_resource_logs is (obviously) only available for logs. This means
you could not use the same receiver for multiple signals unless they
each used the same encoding.

To address both of these issues we introduce signal-specific
configuration: `logs::topic`, `metrics::topic`, `traces::topic`,
`logs::encoding`, `metrics::encoding`, and `traces::encoding`.

The existing `topic` and `encoding` configuration have been deprecated.
If the new fields are set, they will take precedence; otherwise if the
deprecated fields are set they will be used. The defaults have not
changed.

#### Link to tracking issue

Fixes #32735

#### Testing

Unit tests added.

#### Documentation

README updated.

Author: axw

.chloggen/kafkareceiver-persignal-config.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: deprecation
+
+# 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 signal-specific topic and encoding config, deprecate existing topic/encoding config.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32735]
+
+# (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: [user]

receiver/kafkareceiver/README.md

@@ -13,9 +13,7 @@
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->
 
-Kafka receiver receives traces, metrics, and logs from Kafka. Message payload encoding is configurable.
-
-Note that metrics and logs only support OTLP.
+Kafka receiver receives telemetry data from Kafka, with configurable topics and encodings.
 
 If used in conjunction with the `kafkaexporter` configured with `include_metadata_keys`. The Kafka receiver will also propagate the Kafka headers to the downstream pipeline, giving access to the rest of the pipeline to arbitrary metadata keys and values.
 
@@ -28,20 +26,19 @@ 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
-- `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. 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.
-  - `otlp_json`: the payload is deserialized to `ExportTraceServiceRequest` `ExportLogsServiceRequest` or `ExportMetricsServiceRequest` respectively using JSON encoding.
-  - `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`.
-  - `zipkin_proto`: the payload is deserialized into a list of Zipkin proto spans.
-  - `zipkin_json`: the payload is deserialized into a list of Zipkin V2 JSON spans.
-  - `zipkin_thrift`: the payload is deserialized into a list of Zipkin Thrift spans.
-  - `raw`: (logs only) the payload's bytes are inserted as the body of a log record.
-  - `text`: (logs only) the payload are decoded as text and inserted as the body of a log record. By default, it uses UTF-8 to decode. You can use `text_<ENCODING>`, like `text_utf-8`, `text_shift_jis`, etc., to customize this behavior.
-  - `json`: (logs only) the payload is decoded as JSON and inserted as the body of a log record.
-  - `azure_resource_logs`: (logs only) the payload is converted from Azure Resource Logs format to OTel format.
+- `logs`
+  - `topic` (default = otlp\_logs): The name of the Kafka topic from which to consume logs.
+  - `encoding` (default = otlp\_proto): The encoding for the Kafka topic. See [Supported encodings](#supported-encodings).
+- `metrics`
+  - `topic` (default = otlp\_metrics): The name of the Kafka topic from which to consume metrics.
+  - `encoding` (default = otlp\_proto): The encoding for the Kafka topic. See [Supported encodings](#supported-encodings).
+- `traces`
+  - `topic` (default = otlp\_spans): The name of the Kafka topic from which to consume traces.
+  - `encoding` (default = otlp\_proto): The encoding for the Kafka topic. See [Supported encodings](#supported-encodings).
+- `topic` (Deprecated [v0.124.0]: use `logs::topic`, `traces::topic`, or `metrics::topic`).
+   If this is set, it will take precedence over the default value for those fields.
+- `encoding` (Deprecated [v0.124.0]: use `logs::encoding`, `traces::encoding`, or `metrics::encoding`).
+   If this is set, it will take precedence over the default value for those fields.
 - `group_id` (default = otel-collector): The consumer group that receiver will be consuming messages from
 - `client_id` (default = otel-collector): The consumer client ID that receiver will use
 - `initial_offset` (default = latest): The initial offset to use if no offset was previously committed. Must be `latest` or `earliest`.
@@ -97,14 +94,52 @@ The following settings can be optionally configured:
   - `randomization_factor`: A random factor used to calculate next backoff. Randomized interval = RetryInterval * (1 ± RandomizationFactor)
   - `max_elapsed_time`: The maximum amount of time trying to backoff before giving up. If set to 0, the retries are never stopped.
 
-Example:
+### Supported encodings
+
+The Kafka receiver supports encoding extensions, as well as the following built-in encodings.
+
+Available for all signals:
+
+- `otlp_proto`: the payload is decoded as OTLP Protobuf
+- `otlp_json`: the payload is decoded as OTLP JSON
+
+Available only for traces:
+
+- `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`.
+- `zipkin_proto`: the payload is deserialized into a list of Zipkin proto spans.
+- `zipkin_json`: the payload is deserialized into a list of Zipkin V2 JSON spans.
+- `zipkin_thrift`: the payload is deserialized into a list of Zipkin Thrift spans.
+
+Available only for logs:
+
+- `raw`: the payload's bytes are inserted as the body of a log record.
+- `text`: the payload are decoded as text and inserted as the body of a log record. By default, it uses UTF-8 to decode. You can use `text_<ENCODING>`, like `text_utf-8`, `text_shift_jis`, etc., to customize this behavior.
+- `json`: the payload is decoded as JSON and inserted as the body of a log record.
+- `azure_resource_logs`: the payload is converted from Azure Resource Logs format to OTel format.
+
+### Message header propagation
+
+The Kafka receiver will extract Kafka message headers and include them as request metadata (context).
+This metadata can then be used throughout the pipeline, for example to set attributes using the
+[attributes processor](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/attributesprocessor/README.md).
+
+### Example configurations
+
+#### Minimal configuration
+
+By default, the receiver does not require any configuration. With the following configuration,
+the receiver will consume messages from the default topics from localhost:9092 using the
+`otlp_proto` encoding:
 
 ```yaml
 receivers:
   kafka:
-    protocol_version: 2.0.0
 ```
-Example of connecting to kafka using SASL and TLS:
+#### TLS and authentication
+
+In this example the receiver is configured to connect to Kafka using TLS for encryption,
+and SASL/SCRAM for authentication:
 
 ```yaml
 receivers:
@@ -116,39 +151,31 @@ receivers:
         password: "secret"
         mechanism: "SCRAM-SHA-512"
 ```
-Example of header extraction:
+
+#### Header extraction
+
+In addition to propagating Kafka message headers as metadata as described above in
+[Message header propagation](#message-header-propagation), the Kafka receiver can also
+be configured to extract and attach specific headers as resource attributes. e.g.
 
 ```yaml
 receivers:
   kafka:
-    topic: test
     header_extraction: 
       extract_headers: true
       headers: ["header1", "header2"]
 ```
 
-- If we feed following kafka record to `test` topic and use above configs: 
-```yaml
-{
-  event: Hello,
-  headers: {
-    header1: value1,
-    header2: value2,
-  }
-}
+If we produce a Kafka message with headers "header1: value1" and "header2: value2"
+with the above configuration, the receiver will attach these headers as resource
+attributes with the prefix "kafka.header.", i.e.
+
 ```
-we will get a log record in collector similar to: 
-```yaml
-{
-  ...
-  body: Hello,
-  resource: {
-    kafka.header.header1: value1,
-    kafka.header.header2: value2,
-  },
-  ...
+"resource": {
+  "attributes": {
+    "kafka.header.header1": "value1",
+    "kafka.header.header2": "value2",
+  }
 }
+...
 ```
-
-- Here you can see the kafka record header `header1` and `header2` being added to resource attribute.
-- Every **matching** kafka header key is prefixed with `kafka.header` string and attached to resource attributes.

receiver/kafkareceiver/config.go

@@ -6,6 +6,7 @@ package kafkareceiver // import "github.com/open-telemetry/opentelemetry-collect
 import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configretry"
+	"go.opentelemetry.io/collector/confmap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka/configkafka"
 )
@@ -17,22 +18,97 @@ type Config struct {
 	configkafka.ClientConfig   `mapstructure:",squash"`
 	configkafka.ConsumerConfig `mapstructure:",squash"`
 
-	// The name of the kafka topic to consume from (default "otlp_spans" for traces, "otlp_metrics" for metrics, "otlp_logs" for logs)
+	// Logs holds configuration about how logs should be consumed.
+	Logs TopicEncodingConfig `mapstructure:"logs"`
+
+	// Metrics holds configuration about how metrics should be consumed.
+	Metrics TopicEncodingConfig `mapstructure:"metrics"`
+
+	// Traces holds configuration about how traces should be consumed.
+	Traces TopicEncodingConfig `mapstructure:"traces"`
+
+	// Topic holds the name of the Kafka topic from which to consume data.
+	//
+	// Topic has no default. If explicitly specified, it will take precedence
+	// over the default values of Logs.Topic, Traces.Topic, and Metrics.Topic.
+	//
+	// Deprecated [v0.124.0]: Use Logs.Topic, Traces.Topic, and Metrics.Topic.
 	Topic string `mapstructure:"topic"`
 
-	// Encoding of the messages (default "otlp_proto")
+	// Encoding holds the expected encoding of messages (default "otlp_proto")
+	//
+	// Encoding has no default. If explicitly specified, it will take precedence
+	// over the default values of Logs.Encoding, Traces.Encoding, and
+	// Metrics.Encoding.
+	//
+	// Deprecated [v0.124.0]: Use Logs.Encoding, Traces.Encoding, and
+	// Metrics.Encoding.
 	Encoding string `mapstructure:"encoding"`
 
-	// Controls the way the messages are marked as consumed
+	// MessageMarking controls the way the messages are marked as consumed.
 	MessageMarking MessageMarking `mapstructure:"message_marking"`
 
-	// Extract headers from kafka records
+	// HeaderExtraction controls extraction of headers from Kafka records.
 	HeaderExtraction HeaderExtraction `mapstructure:"header_extraction"`
 
-	// In case of some errors returned by the next consumer, the receiver will wait and retry the failed message
+	// ErrorBackoff controls backoff/retry behavior when the next consumer
+	// returns an error.
 	ErrorBackOff configretry.BackOffConfig `mapstructure:"error_backoff"`
 }
 
+func (c *Config) Unmarshal(conf *confmap.Conf) error {
+	if err := conf.Unmarshal(c); err != nil {
+		return err
+	}
+	// Check if deprecated fields have been explicitly set,
+	// in which case they should be used instead of signal-
+	// specific defaults.
+	var zeroConfig Config
+	if err := conf.Unmarshal(&zeroConfig); err != nil {
+		return err
+	}
+	if c.Topic != "" {
+		if zeroConfig.Logs.Topic == "" {
+			c.Logs.Topic = c.Topic
+		}
+		if zeroConfig.Metrics.Topic == "" {
+			c.Metrics.Topic = c.Topic
+		}
+		if zeroConfig.Traces.Topic == "" {
+			c.Traces.Topic = c.Topic
+		}
+	}
+	if c.Encoding != "" {
+		if zeroConfig.Logs.Encoding == "" {
+			c.Logs.Encoding = c.Encoding
+		}
+		if zeroConfig.Metrics.Encoding == "" {
+			c.Metrics.Encoding = c.Encoding
+		}
+		if zeroConfig.Traces.Encoding == "" {
+			c.Traces.Encoding = c.Encoding
+		}
+	}
+	return conf.Unmarshal(c)
+}
+
+// TopicEncodingConfig holds signal-specific topic and encoding configuration.
+type TopicEncodingConfig struct {
+	// Topic holds the name of the Kafka topic from which messages of the
+	// signal type should be consumed.
+	//
+	// The default depends on the signal type:
+	//  - "otlp_spans" for traces
+	//  - "otlp_metrics" for metrics
+	//  - "otlp_logs" for logs
+	Topic string `mapstructure:"topic"`
+
+	// Encoding holds the expected encoding of messages for the signal type
+	//
+	// Defaults to "otlp_proto".
+	Encoding string `mapstructure:"encoding"`
+}
+
 type MessageMarking struct {
 	// If true, the messages are marked after the pipeline execution
 	After bool `mapstructure:"after"`

receiver/kafkareceiver/config_test.go

@@ -46,8 +46,65 @@ func TestLoadConfig(t *testing.T) {
 					config.GroupID = "the_group_id"
 					return config
 				}(),
-				Topic:    "spans",
-				Encoding: "otlp_proto",
+				Logs: TopicEncodingConfig{
+					Topic:    "spans",
+					Encoding: "otlp_proto",
+				},
+				Metrics: TopicEncodingConfig{
+					Topic:    "spans",
+					Encoding: "otlp_proto",
+				},
+				Traces: TopicEncodingConfig{
+					Topic:    "spans",
+					Encoding: "otlp_proto",
+				},
+				Topic: "spans",
+				ErrorBackOff: configretry.BackOffConfig{
+					Enabled: false,
+				},
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "legacy_topic"),
+			expected: &Config{
+				ClientConfig:   configkafka.NewDefaultClientConfig(),
+				ConsumerConfig: configkafka.NewDefaultConsumerConfig(),
+				Logs: TopicEncodingConfig{
+					Topic:    "legacy_topic",
+					Encoding: "otlp_proto",
+				},
+				Metrics: TopicEncodingConfig{
+					Topic:    "metrics_topic",
+					Encoding: "otlp_proto",
+				},
+				Traces: TopicEncodingConfig{
+					Topic:    "legacy_topic",
+					Encoding: "otlp_proto",
+				},
+				Topic: "legacy_topic",
+				ErrorBackOff: configretry.BackOffConfig{
+					Enabled: false,
+				},
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "legacy_encoding"),
+			expected: &Config{
+				ClientConfig:   configkafka.NewDefaultClientConfig(),
+				ConsumerConfig: configkafka.NewDefaultConsumerConfig(),
+				Logs: TopicEncodingConfig{
+					Topic:    "otlp_logs",
+					Encoding: "legacy_encoding",
+				},
+				Metrics: TopicEncodingConfig{
+					Topic:    "otlp_metrics",
+					Encoding: "metrics_encoding",
+				},
+				Traces: TopicEncodingConfig{
+					Topic:    "otlp_spans",
+					Encoding: "legacy_encoding",
+				},
+				Encoding: "legacy_encoding",
 				ErrorBackOff: configretry.BackOffConfig{
 					Enabled: false,
 				},
@@ -82,8 +139,18 @@ func TestLoadConfig(t *testing.T) {
 					config.HeartbeatInterval = 15 * time.Second
 					return config
 				}(),
-				Topic:    "logs",
-				Encoding: "direct",
+				Logs: TopicEncodingConfig{
+					Topic:    "logs",
+					Encoding: "direct",
+				},
+				Metrics: TopicEncodingConfig{
+					Topic:    "otlp_metrics",
+					Encoding: "otlp_proto",
+				},
+				Traces: TopicEncodingConfig{
+					Topic:    "otlp_spans",
+					Encoding: "otlp_proto",
+				},
 				ErrorBackOff: configretry.BackOffConfig{
 					Enabled:         true,
 					InitialInterval: 1 * time.Second,