kafkaexporter

Kafka Exporter

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

Kafka exporter exports logs, metrics, and traces to Kafka. This exporter uses a synchronous producer that blocks and does not batch messages, therefore it should be used with batch and queued retry processors for higher throughput and resiliency. Message payload encoding is configurable.

Configuration settings

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

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.
• client_id (default = "otel-collector"): The client ID to configure the Kafka client with. The client ID will be used for all produce requests.
• conn_idle_timeout (default = 9m): The time after which idle connections to Kafka brokers are not reused and may be closed.
• logs
  • topic (default = otlp_logs): The name of the Kafka topic to which logs will be exported.
  • encoding (default = otlp_proto): The encoding for logs. See Supported encodings.
  • topic_from_metadata_key (default = ""): The name of the metadata key whose value should be used as the message's topic. Useful to dynamically produce to topics based on request inputs. It takes precedence over topic_from_attribute and topic settings.
  • message_key_from_metadata_key (default = ""): The name of the metadata key whose value should be used as the Kafka record key for log messages. If the metadata key is absent or empty, the record key is left nil. Mutually exclusive with partition_logs_by_resource_attributes and partition_logs_by_trace_id. See Message Key for details.
• metrics
  • topic (default = otlp_metrics): The name of the Kafka topic to publish metrics to.
  • encoding (default = otlp_proto): The encoding for metrics. See Supported encodings.
  • topic_from_metadata_key (default = ""): The name of the metadata key whose value should be used as the message's topic. Useful to dynamically produce to topics based on request inputs. It takes precedence over topic_from_attribute and topic settings.
  • message_key_from_metadata_key (default = ""): The name of the metadata key whose value should be used as the Kafka record key for metric messages. If the metadata key is absent or empty, the record key is left nil. Mutually exclusive with partition_metrics_by_resource_attributes. See Message Key for details.
• traces
  • topic (default = otlp_spans): The name of the Kafka topic to publish traces to.
  • encoding (default = otlp_proto): The encoding for traces. See Supported encodings.
  • topic_from_metadata_key (default = ""): The name of the metadata key whose value should be used as the message's topic. Useful to dynamically produce to topics based on request inputs. It takes precedence over topic_from_attribute and topic settings.
  • message_key_from_metadata_key (default = ""): The name of the metadata key whose value should be used as the Kafka record key for trace messages. If the metadata key is absent or empty, the record key is left nil. Mutually exclusive with partition_traces_by_id. See Message Key for details.
• topic_from_attribute (default = ""): Specify the resource attribute whose value should be used as the message's topic. See Destination Topic below for more details.
• include_metadata_keys (default = []): Specifies a list of metadata keys to propagate as Kafka message headers. If one or more keys aren't found in the metadata, they are ignored. When sending_queue::batch is enabled, sending_queue::batch::partition::metadata_keys must be configured and include all values configured in include_metadata_keys.
• record_headers (default = {}): Specifies a map of key/value pairs to set as static headers on every outgoing Kafka record.
• partition_traces_by_id (default = false): configures the exporter to include the trace ID as the message key in trace messages sent to kafka. Please note: this setting does not have any effect on Jaeger encoding exporters since Jaeger exporters include trace ID as the message key by default.
• partition_metrics_by_resource_attributes (default = false) configures the exporter to include the hash of sorted resource attributes as the message partitioning key in metric messages sent to kafka.
• partition_logs_by_resource_attributes (default = false) configures the exporter to include the hash of sorted resource attributes as the message partitioning key in log messages sent to kafka.
• partition_logs_by_trace_id (default = false): configures the exporter to partition log messages by trace ID, if the log record has one associated. Note: partition_logs_by_resource_attributes and partition_logs_by_trace_id are mutually exclusive, and enabling both will lead to an error.
• record_partitioner: Configures the Kafka record partitioning strategy. At most one option may be set. If multiple partitioners are configured, an error will be returned. If none is configured, the default is sticky_key with a Sarama-compatible hasher (preserving previous behavior).
  • sticky_key: Uses a sticky-key partitioner.
    • hasher: The hash algorithm used for key-based partition assignment.
      • sarama_compat (default): Uses Sarama-compatible FNV-1a hashing.
      • murmur2: Uses Murmur2 hashing
  • round_robin: Distributes records evenly across all available partitions in round-robin order.
  • least_backup: Routes each record to the partition with the fewest buffered (in-flight) records.
  • extension: The component ID of a custom partitioner extension. When set, partitioning is delegated to the specified extension.
• tls: see TLS Configuration Settings for the full set of available options. Set to tls: insecure: false explicitly when using AWS_MSK_IAM_OAUTHBEARER as the authentication method.
• 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)
    • version (default = 0): The SASL protocol version to use (0 or 1)
    • 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
  • refresh_interval (default = 10m): The refreshInterval controls the frequency at which cluster metadata is refreshed in the background.
• timeout (default = 5s): Time to wait per individual attempt to produce data to Kafka.
• retry_on_failure
  • enabled (default = true)
  • initial_interval (default = 5s): Time to wait after the first failure before retrying; ignored if enabled is false
  • randomization_factor (default = 0.5): Is the random factor used to calculate the next backoffs.
  • multiplier (default = 1.5): Is the value multiplied by the backoff interval bounds.
  • max_interval (default = 30s): Is the upper bound on backoff; ignored if enabled is false
  • max_elapsed_time (default = 300s): Is the maximum amount of time spent trying to send a batch; ignored if enabled is false
• sending_queue: see Exporter Helper for the full set of available options.
  • enabled (default = true)
  • num_consumers (default = 10): Number of consumers that dequeue batches; ignored if enabled is false
  • queue_size (default = 1000): Maximum number of batches kept in memory before dropping data; ignored if enabled is false; User should calculate this as num_seconds * requests_per_second where:
    • num_seconds is the number of seconds to buffer in case of a backend outage
    • requests_per_second is the average number of requests per seconds.
• producer
  • max_message_bytes (default = 1000000) the maximum permitted size of a message in bytes, calculated before compression.
  • max_broker_write_bytes (default = 104857600) the maximum bytes the producer will write to a broker in a single request. Must be greater than or equal to max_message_bytes. Increase this when raising max_message_bytes above 100 MiB.
  • required_acks (default = 1) controls when a message is regarded as transmitted. https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks
  • compression (default = 'none') the compression used when producing messages to kafka. The options are: none, gzip, snappy, lz4, and zstd https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#compression-type
  • compression_params
    • level (default = -1) the compression level used when producing messages to kafka.
    • The following are valid combinations of compression and level
      • gzip
        • BestSpeed: 1
        • BestCompression: 9
        • DefaultCompression: -1
      • zstd
        • SpeedFastest: 1
        • SpeedDefault: 3
        • SpeedBetterCompression: 6
        • SpeedBestCompression: 11
      • lz4 Only supports fast level
      • snappy No compression levels supported yet
  • flush_max_messages (default = 10000) The maximum number of messages the producer will send in a single broker request.
  • allow_auto_topic_creation (default = true) whether the broker is allowed to automatically create topics when they are referenced but do not already exist.
  • linger: (default = 10ms) How long individual topic partitions will linger waiting for more records before triggering a request to be built.

Supported encodings

The Kafka exporter supports encoding extensions, as well as the following built-in encodings.

Available for all signals:

• otlp_proto: data is encoded as OTLP Protobuf
• otlp_json: data is encoded as OTLP JSON

Available only for traces:

• jaeger_proto: the payload is serialized to a single Jaeger proto Span, and keyed by TraceID.
• jaeger_json: the payload is serialized to a single Jaeger JSON Span using jsonpb, and keyed by TraceID.
• zipkin_proto: the payload is serialized to Zipkin v2 proto Span.
• zipkin_json: the payload is serialized to Zipkin v2 JSON Span.

Available only for logs:

• raw: if the log record body is a byte array, it is sent as is. Otherwise, it is serialized to JSON. Resource and record attributes are discarded.

Example configuration

Example configuration:

exporters:
  kafka:
    brokers:
      - localhost:9092
    record_headers:
      my-custom-header: "my-custom-value"
      another-header: "another-value"

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

exporters:
  kafka:
    brokers: ["localhost:9092"]
    tls:
      insecure: false
    auth:
      sasl:
        mechanism: OAUTHBEARER
        oauthbearer_token_source: oauth2client

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

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.

Destination Topic

The destination topic can be defined in a few different ways and takes priority in the following order:

1. When <signal>::topic_from_metadata_key is set to use a key from the request metadata, the value of this key is used as the signal specific topic.
2. Otherwise, if topic_from_attribute is configured, and the corresponding attribute is found on the ingested data, the value of this attribute is used.
3. If a prior component in the collector pipeline sets the topic on the context via the topic.WithTopic function (from the github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic package), the value set in the context is used.
4. Finally, the <signal>::topic configuration is used for the signal-specific destination topic.

Message Key

The Kafka record key can be set in the following ways, in order of precedence:

1. When <signal>::message_key_from_metadata_key is configured and the named metadata key is present and non-empty, its value is used as the record key. This is intended to be used together with an upstream processor that evaluates OTTL expressions and stores the result in request metadata.
2. When one of the partition_* flags is set (partition_traces_by_id, partition_metrics_by_resource_attributes, partition_logs_by_resource_attributes, or partition_logs_by_trace_id), the record key is derived from the signal data. These flags are mutually exclusive with message_key_from_metadata_key for the same signal.
3. For Jaeger encodings (jaeger_proto, jaeger_json), the marshaler always keys records by the trace ID.
4. Otherwise the record key is nil and the configured record_partitioner strategy determines which partition receives the record.

Partitioning Kafka Records

The exporter supports multiple strategies to control how records are distributed across kafka partitions within a topic.

Available strategies for partitioning are sticky_key, sticky, round_robin, least_backup and extension

Using custom partitioner

The Kafka exporter allows you to define a custom partitioning strategy via an extension. A sample config for custom partitioner would look like:

exporters:
  kafka:
    brokers:
      - localhost:9092
    record_partitioner:
      extension: my_custom_partitioner
  
extensions:
  my_custom_partitioner:
    # your extension-specific configuration here

# rest of the pipeline config

Use custom partitioner if:

• Built-in strategies (round_robin, least_backup, etc.) don’t fit your needs
• You require domain-specific routing logic

[!NOTE] The custom partitioner extension must implement the RecordPartitionerExtension interface (see partitioner.go).