Skip to main content

​
Kafka Receiver

Status Available in: core, contrib Maintainers: @pavolloffay, @MovieStoreGuy, @paulojmdias Source: opentelemetry-collector-contrib

​
Supported Telemetry

Logs Metrics Traces

​
Overview

​
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 any of the topics have the ^ prefix, regex consuming will be enabled and exclude_topics can be used (see Regex topic patterns with exclusions for an example).
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
    • 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_topics (default = []): When at least one entry in topics is a regex pattern (prefix with ^), topics matching any of these regular expressions are excluded.
  • metrics
    • 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_topics (default = []): When at least one entry in topics is a regex pattern (prefix with ^), topics matching any of these regular expressions are excluded.
  • traces
    • 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_topics (default = []): When at least one entry in topics is a regex pattern (prefix with ^), topics matching any of these regular expressions are excluded.
  • profiles
    • 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_topics (default = []): When at least one entry in topics is a regex pattern (prefix with ^), topics matching any of these regular expressions are excluded.
  • 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 (deprecated): No longer has any effect. Metadata is only fetched for the topics the client consumes from.
    • refresh_interval (default = 10m): The refreshInterval controls the frequency at which cluster metadata is refreshed in the background.
    • retry
      • max (default = 20): The number of times to retry a retriable request. Applies to consumer group and offset commit/fetch requests. It does not apply to fetch requests, nor to the client’s internal metadata refresh, which caps itself at 3 retries.
      • backoff (default = 250ms): The minimum time to wait before retrying a request. Each successive retry doubles the wait, with jitter applied, capped at max(5s, backoff). Unlike max, this applies to every retry path, including fetches.
  • 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.
  • partition_processing:
    • independent (default = false): Process each assigned topic partition sequentially in its own worker so a blocked partition does not block polling healthy partitions. Requires autocommit.enable to be true.
    • max_buffered_batches (default = 1): Maximum number of fetched batches waiting for each partition worker. Must be greater than zero when independent processing is enabled.
  • 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.

​
Trace context propagation

The receiver extracts the trace context from record headers, using the propagators configured in service::telemetry::traces::propagators. The receiver span starts a new trace and links to the extracted trace context. Propagators are only installed when service::telemetry::traces also has processors configured, so without them, no trace context is extracted.

​
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:

​
TLS and authentication

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

​
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:
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.
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.

​
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] exclude_topics only takes effect when at least one entry in topics is a regex pattern (prefixed with ^); the receiver fails config validation otherwise. Entries in exclude_topics are regular expressions matched unanchored, so exclude_topics: ["test"] also excludes a topic named prod-test-1. Prefix them with ^ (and anchor with $ where appropriate), as in the example below.
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-

​
Independent partition processing

NOTE: Independent partition processing requires autocommit.enable: true. The receiver rejects configurations that combine independent processing with manual commits.
Independent partition processing keeps records ordered within each partition while allowing other assigned partitions to continue when one downstream consumer is blocked. Each partition has a bounded mailbox. A full mailbox pauses only that partition and resumes it when capacity becomes available. The receiver creates one worker and mailbox per assigned partition. max_buffered_batches limits the number of fetched batches waiting in each mailbox, not the number of records or workers. Resource usage grows with the number of assigned partitions and max_buffered_batches. Increasing mailbox capacity allows more fetched data to remain in memory while waiting for processing. When a partition is revoked, its worker is cancelled and queued batches are discarded. The next owner fetches those records again from the committed offset, so standard Kafka at-least-once delivery and possible duplication still apply.

​
Attributes

​
Configuration

​
Example Configuration

Last generated: 2026-10-05