Documentation Grafana Cloud Instrument and send data Grafana Alloy Reference Components otelcol otelcol.receiver.kafka

otelcol.receiver.kafka

otelcol.receiver.kafka accepts telemetry data from a Kafka broker and forwards it to other otelcol.* components.

You can specify multiple otelcol.receiver.kafka components by giving them different labels.

Usage

alloy Copy

otelcol.receiver.kafka "<LABEL>" {
  brokers          = ["<BROKER_ADDR>"]
  protocol_version = "<PROTOCOL_VERSION>"

  output {
    metrics = [...]
    logs    = [...]
    traces  = [...]
  }
}

Arguments

You can use the following arguments with otelcol.receiver.kafka:

For max_fetch_size, the value 0 means no limit.

initial_offset must be either "latest" or "earliest".

The group_rebalance_strategy argument determines how Kafka distributes topic partitions among the consumers in the group during rebalances. Supported strategies are:

• range: This strategy assigns partitions to consumers based on a range. It aims to distribute partitions evenly across consumers, but it can lead to uneven distribution if the number of partitions is not a multiple of the number of consumers. For more information, refer to the Kafka RangeAssignor documentation, see RangeAssignor.
• roundrobin: This strategy assigns partitions to consumers in a round-robin fashion. It ensures a more even distribution of partitions across consumers, especially when the number of partitions is not a multiple of the number of consumers. For more information, refer to the Kafka RoundRobinAssignor documentation, see RoundRobinAssignor.
• sticky: This strategy aims to maintain the same partition assignments during rebalances as much as possible. It minimizes the number of partition movements, which can be beneficial for stateful consumers. For more information, refer to the Kafka StickyAssignor documentation, see StickyAssignor.

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.

• If group_instance_id is 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 group_instance_id is 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.

Blocks

You can use the following blocks with otelcol.receiver.kafka:

The > symbol indicates deeper levels of nesting. For example, authentication > tls refers to a tls block defined inside an authentication block.

output

logs

The logs block configures how to receive logs from Kafka brokers.

metrics

The metrics block configures how to receive metrics from Kafka brokers.

traces

The traces block configures how to receive traces from Kafka brokers.

authentication

kerberos

plaintext

sasl

aws_msk

tls

The tls block configures TLS settings used for connecting to the Kafka brokers. If the tls block isn’t provided, TLS won’t be used for communication.

tpm

The tpm block configures retrieving the TLS key_file from a trusted device.

The following arguments are supported:

Name	Type	Description	Default	Required
auth	string	The authorization value used to authenticate the TPM device.	""	no
enabled	bool	Load the tls.key_file from TPM.	false	no
owner_auth	string	The owner authorization value used to authenticate the TPM device.	""	no
path	string	Path to the TPM device or Unix domain socket.	""	no

The trusted platform module (TPM) configuration can be used for loading TLS key from TPM. Currently only TSS2 format is supported.

The path attribute is not supported on Windows.

Example

alloy Copy

otelcol.example.component "<LABEL>" {
    ...
    tls {
        ...
        key_file = "my-tss2-key.key"
        tpm {
            enabled = true
            path = "/dev/tpmrm0"
        }
    }
}

In the above example, the private key my-tss2-key.key in TSS2 format will be loaded from the TPM device /dev/tmprm0.

autocommit

The autocommit block configures how to automatically commit updated topic offsets back to the Kafka brokers.

The following arguments are supported:

debug_metrics

error_backoff

The error_backoff block configures how failed requests to Kafka are retried.

The following arguments are supported:

When enabled is true, failed batches are retried after a given interval. The initial_interval argument specifies how long to wait before the first retry attempt. If requests continue to fail, the time to wait before retrying increases by the factor specified by the multiplier argument, which must be greater than 1.0. The max_interval argument specifies the upper bound of how long to wait between retries.

The randomization_factor argument is useful for adding jitter between retrying Alloy instances. If randomization_factor is greater than 0, the wait time before retries is multiplied by a random factor in the range [ I - randomization_factor * I, I + randomization_factor * I], where I is the current interval.

If a batch hasn’t been sent successfully, it’s discarded after the time specified by max_elapsed_time elapses. If max_elapsed_time is set to "0s", failed requests are retried forever until they succeed.

header_extraction

The header_extraction block configures how to extract headers from Kafka records.

The following arguments are supported:

Regular expressions aren’t allowed in the headers argument. Only exact matching will be performed.

message_marking

The message_marking block configures when Kafka messages are marked as read.

The following arguments are supported:

By default, a Kafka message is marked as read immediately after it’s retrieved from the Kafka broker. If the after_execution argument is true, messages are only read after the telemetry data is forwarded to components specified in the output block.

When after_execution is true, messages are only marked as read when they’re decoded successfully and components where the data was forwarded didn’t return an error. If the include_unsuccessful argument is true, messages are marked as read even if decoding or forwarding failed. Setting include_unsuccessful has no effect if after_execution is false.

metadata

retry

Exported fields

otelcol.receiver.kafka doesn’t export any fields.

Supported encodings

otelcol.receiver.kafka 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

otelcol.receiver.kafka 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 otelcol.processor.attributes.

Component health

otelcol.receiver.kafka is only reported as unhealthy if given an invalid configuration.

Debug information

otelcol.receiver.kafka doesn’t expose any component-specific debug information.

Example

This example forwards read telemetry data through a batch processor before finally sending it to an OTLP-capable endpoint:

alloy Copy

otelcol.receiver.kafka "default" {
  brokers          = ["localhost:9092"]
  protocol_version = "2.0.0"

  output {
    metrics = [otelcol.processor.batch.default.input]
    logs    = [otelcol.processor.batch.default.input]
    traces  = [otelcol.processor.batch.default.input]
  }
}

otelcol.processor.batch "default" {
  output {
    metrics = [otelcol.exporter.otlp.default.input]
    logs    = [otelcol.exporter.otlp.default.input]
    traces  = [otelcol.exporter.otlp.default.input]
  }
}

otelcol.exporter.otlp "default" {
  client {
    endpoint = sys.env("OTLP_ENDPOINT")
  }
}

Compatible components

otelcol.receiver.kafka can accept arguments from the following components:

• Components that export OpenTelemetry otelcol.Consumer