kafkareceiver

Kafka Receiver

The 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.

Status
Stability	development: profiles
	beta: metrics, logs, traces
Distributions	core, contrib
Issues
Code coverage
Code Owners	@pavolloffay, @MovieStoreGuy, @axw, @paulojmdias

Getting Started

[!NOTE] The Kafka receiver uses the franz-go client library, which provides better performance and support for modern Kafka features.

The franz-go client supports directly consuming from multiple topics by specifying a regex expression. To enable this feature, prefix your topic with the ^ character. This is identical to how the librdkafka client works.

If you use the ^ prefix, in the deprecated topic setting, if any of the topics have the ^ prefix, regex consuming will be enabled.

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
• logs
  • topic (Deprecated [v0.142.0]: use topics) (default = otlp_logs): If this is set, it will take precedence over default value of topics
  • topics (default = otlp_logs): List of kafka topics from which to consume logs
  • encoding (default = otlp_proto): The encoding for the Kafka topic. See Supported encodings.
  • exclude_topic (Deprecated [v0.142.0]: use exclude_topics) (default = ""): If this is set, it will take precedence over default value of exclude_topics
  • exclude_topics (default = ""): When using regex topic patterns (prefix with ^), this regex pattern excludes matching topics.
• metrics
  • topic (Deprecated [v0.142.0]: use topics) (default = otlp_metrics): If this is set, it will take precedence over default value of topics
  • topics (default = otlp_metrics): List of Kafka topic from which to consume metrics.
  • encoding (default = otlp_proto): The encoding for the Kafka topic. See Supported encodings.
  • exclude_topic (Deprecated [v0.142.0]: use exclude_topics) (default = ""): If this is set, it will take precedence over default value of exclude_topics
  • exclude_topics (default = ""): When using regex topic patterns (prefix with ^), this regex pattern excludes matching topics.
• traces
  • topic (Deprecated [v0.142.0]: use topics) (default = otlp_spans): If this is set, it will take precedence over default value of topics
  • topics (default = otlp_spans): List of Kafka topic from which to consume traces.
  • encoding (default = otlp_proto): The encoding for the Kafka topic. See Supported encodings.
  • exclude_topic (Deprecated [v0.142.0]: use exclude_topics) (default = ""): If this is set, it will take precedence over default value of exclude_topics
  • exclude_topics (default = ""): When using regex topic patterns (prefix with ^), this regex pattern excludes matching topics.
• profiles
  • topic (Deprecated [v0.142.0]: use topics) (default = otlp_profiles): If this is set, it will take precedence over default value of topics
  • topics (default = otlp_profiles): List of Kafka topic from which to consume profiles.
  • encoding (default = otlp_proto): The encoding for the Kafka topic. See Supported encodings.
  • exclude_topic (Deprecated [v0.142.0]: use exclude_topics) (default = ""): If this is set, it will take precedence over default value of exclude_topics
  • exclude_topics (default = ""): When using regex topic patterns (prefix with ^), this regex pattern excludes matching topics.
• 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
• rack_id (default = ""): The rack identifier for this client. When set and brokers are configured with a rack-aware replica selector, the client will prefer fetching from the closest replica.
• use_leader_epoch (default = true): (Experimental) When enabled, the consumer uses the leader epoch returned by brokers (KIP-320) to detect log truncation. Setting this to false clears the leader epoch from fetch offsets, disabling KIP-320. Disabling can improve compatibility with brokers that don’t fully support leader epochs (e.g., Azure Event Hubs), at the cost of losing automatic log-truncation safety.
• conn_idle_timeout (default = 9m): The time after which idle connections to Kafka brokers are not reused and may be closed.
• initial_offset (default = latest): The initial offset to use if no offset was previously committed. Must be latest or earliest.
• session_timeout (default = 10s): The request timeout for detecting client failures when using Kafka’s group management facilities.
• heartbeat_interval (default = 3s): The expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities.
• group_rebalance_strategies: Ordered list of strategies to advertise to Kafka. Kafka selects the first advertised strategy supported by every member of the consumer group. When omitted, the franz-go default applies, currently cooperative-sticky. This field is mutually exclusive with group_rebalance_strategy; setting both fails validation. This can be used for rolling migrations between assignment protocols. Built-in values are:
  • range: Assigns partitions per topic based on a contiguous range. See RangeAssignor.
  • roundrobin: Assigns partitions across all topics in a round-robin fashion. See RoundRobinAssignor.
  • sticky: Minimises partition movement across rebalances. See StickyAssignor.
  • cooperative-sticky: Like sticky, but uses cooperative (incremental) rebalancing to avoid a full stop-the-world rebalance. See CooperativeStickyAssignor.
  • Any other value is interpreted as the component ID of a registered extension that implements kgo.GroupBalancer, which allows a custom partition-assignment strategy to be plugged in.
• group_rebalance_strategy (default = cooperative-sticky, Deprecated [v0.154.0]: use group_rebalance_strategies instead): The strategy used to assign partitions to consumers within a consumer group during rebalances. Accepts the same values as group_rebalance_strategies. Setting this value logs a deprecation warning on startup. It is mutually exclusive with group_rebalance_strategies; setting both fails validation.
• group_instance_id: A unique identifier for the consumer instance within a consumer group.
  • If set to a non-empty string, the consumer is treated as a static member of the group. This means that the consumer will maintain its partition assignments across restarts and rebalances, as long as it rejoins the group with the same group_instance_id.
  • If set to an empty string (or not set), the consumer is treated as a dynamic member. In this case, the consumer's partition assignments may change during rebalances.
  • Using a group_instance_id is useful for stateful consumers or when you need to ensure that a specific consumer instance is always assigned the same set of partitions.
• min_fetch_size (default = 1): The minimum number of message bytes to fetch in a request, defaults to 1 byte.
• max_fetch_size (default = 1048576): The maximum number of message bytes to fetch in a request, defaults to 1MB. Must be greater than or equal to min_fetch_size.
• max_fetch_wait (default = 250ms): The maximum amount of time the broker should wait for min_fetch_size bytes to be available before returning anyway.
• max_partition_fetch_size (default = 1048576): The default number of message bytes to fetch in a request per partition, defaults to 1MB. If a single record batch is larger than this value, the broker will still return it to ensure the consumer can make progress.
• tls: see TLS Configuration Settings for the full set of available options.
• auth
  • plain_text (Deprecated in v0.123.0: use sasl with mechanism set to PLAIN instead.)
    • username: The username to use.
    • password: The password to use
  • sasl
    • username: The username to use.
    • password: The password to use.
    • mechanism: The sasl mechanism to use (SCRAM-SHA-256, SCRAM-SHA-512, AWS_MSK_IAM_OAUTHBEARER, OAUTHBEARER, or PLAIN)
    • aws_msk
      • region: AWS Region in case of AWS_MSK_IAM_OAUTHBEARER mechanism
    • oauthbearer_token_source: The component ID of an authenticator extension that provides OAuth2 tokens (e.g. oauth2client or azure_auth). Required when mechanism is OAUTHBEARER; the extension must be listed under service.extensions.
  • tls (Deprecated in v0.124.0: configure tls at the top level): this is an alias for tls at the top level.
  • kerberos
    • service_name: Kerberos service name
    • realm: Kerberos realm
    • use_keytab: Use of keytab instead of password, if this is true, keytab file will be used instead of password
    • username: The Kerberos username used for authenticate with KDC
    • password: The Kerberos password used for authenticate with KDC
    • config_file: Path to Kerberos configuration. i.e /etc/krb5.conf
    • keytab_file: Path to keytab file. i.e /etc/security/kafka.keytab
    • disable_fast_negotiation: Disable PA-FX-FAST negotiation (Pre-Authentication Framework - Fast). Some common Kerberos implementations do not support PA-FX-FAST negotiation. This is set to false by default.
• metadata
  • full (default = true): Whether to maintain a full set of metadata. When disabled, the client does not make the initial request to broker at the startup.
  • refresh_interval (default = 10m): The refreshInterval controls the frequency at which cluster metadata is refreshed in the background.
  • retry
    • max (default = 3): The number of retries to get metadata
    • backoff (default = 250ms): How long to wait between metadata retries
• autocommit
  • enable: (default = true) Whether or not to auto-commit updated offsets back to the broker
  • interval: (default = 1s) How frequently to commit updated offsets. Ineffective unless auto-commit is enabled
• message_marking:
  • after: (default = false) If true, the messages are marked after the pipeline execution
  • on_error: (default = false) If false, only the successfully processed messages are marked. This applies to non-permanent errors. Note: when error_backoff is enabled, the failed record is automatically retried on the next poll cycle once all retries are exhausted. Without error_backoff, the partition remains paused until a rebalance occurs.
  • on_permanent_error: (default = value of on_error) If false, messages that generate permanent errors are not marked. If true, messages that generate permanent errors are marked. Note: this can block the entire partition in case a message processing returns a permanent error. Permanent errors are not retried via error_backoff, but the uncommitted message will be reprocessed after a rebalance.
• header_extraction:
  • extract_headers (default = false): Allows user to attach header fields to resource attributes in otel pipeline
  • headers (default = []): List of headers they'd like to extract from kafka record. Note: Matching pattern will be exact. Regexes are not supported as of now.
• error_backoff: BackOff configuration in case of errors
  • enabled: (default = false) Whether to enable backoff when next consumers return errors
  • initial_interval: The time to wait after the first error before retrying
  • max_interval: The upper bound on backoff interval between consecutive retries
  • multiplier: The value multiplied by the backoff interval bounds
  • 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.
• telemetry
  • metrics
    • kafka_receiver_records_delay:
      • enabled (default = false) Whether the metric kafka_receiver_records_delay will be reported or not.

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 (Deprecated [v0.149.0]: use azureencodingextension): the payload is converted from Azure Resource Logs format to OTel format.

Message metadata propagation

The Kafka receiver includes the following record metadata as request metadata (context) for each consumed message:

• kafka.topic: the topic the message was consumed from
• kafka.partition: the partition the message was consumed from
• kafka.offset: the offset of the message within the partition

Additionally, all Kafka message headers are included in the request metadata.

This metadata can then be used throughout the pipeline, for example to set attributes using the attributes processor.

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:

receivers:
  kafka:

TLS and authentication

In this example the receiver is configured to connect to Kafka using TLS for encryption, and SASL/SCRAM for authentication:

receivers:
  kafka:
    tls:
    auth:
      sasl:
        username: "user"
        password: "secret"
        mechanism: "SCRAM-SHA-512"

SASL/OAUTHBEARER (OAuth2 token source)

The OAUTHBEARER mechanism delegates token acquisition and refresh to an authenticator extension referenced by oauthbearer_token_source, such as oauth2clientauth:

extensions:
  oauth2client:
    client_id: someclientid
    client_secret: someclientsecret
    token_url: https://example.com/oauth2/default/v1/token

receivers:
  kafka:
    tls:
      insecure: false
    auth:
      sasl:
        mechanism: OAUTHBEARER
        oauthbearer_token_source: oauth2client

service:
  extensions: [oauth2client]
  pipelines:
    logs:
      receivers: [kafka]
      exporters: [debug]

The azureauth extension can also be used as a token source, which supports managed identity, workload identity, and service principal for authenticating against Azure Event Hubs.

Header extraction

In addition to propagating Kafka message metadata as described above in Message metadata propagation, the Kafka receiver can also be configured to extract and attach specific headers as resource attributes. e.g.

receivers:
  kafka:
    header_extraction:
      extract_headers: true
      headers: ["header1", "header2"]

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.

"resource": {
  "attributes": {
    "kafka.header.header1": "value1",
    "kafka.header.header2": "value2",
  }
}
...

Regex topic patterns with exclusions

When using the franz-go client, you can consume from multiple topics using regex patterns and exclude specific topics from consumption. This is useful when you want to consume from a dynamic set of topics but need to filter out certain ones.

Note: Both topic and exclude_topic must use regex patterns (prefix with ^) for exclusion to work. This feature is only available with the franz-go client.

receivers:
  kafka:
    logs:
      topics:
      - "^logs-.*"                   # Consume from all topics matching logs-*
      exclude_topics:
      - "^logs-(test|dev)$"  # Exclude logs-test and logs-dev
    metrics:
      topics:
      - "^metrics-.*"
      exclude_topics:
      - "^metrics-internal-.*$"

In the example above:

• For logs: the receiver will consume from topics like logs-prod, logs-staging, logs-app but will exclude logs-test and logs-dev
• For metrics: the receiver will consume from topics like metrics-app, metrics-infra but will exclude any topics starting with metrics-internal-