Registry /
observability / opentelemetry-instrumentation-aiokafka
Install & Compatibility
Where this runs
tested against v0.65b0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.95 runs
build_error
glibcpy 3.10–3.95 runs
installs and imports cleanly · install 2.9s · import 0.000s · 29MB
27MB installed
● package 27MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
AioKafkaInstrumentor
✓ from opentelemetry.instrumentation.aiokafka import AioKafkaInstrumentor
This quickstart demonstrates how to instrument `aiokafka` producer and consumer using `opentelemetry-instrumentation-aiokafka`. It sets up a basic `TracerProvider` with a `ConsoleSpanExporter` to print traces to the console, then uses `AioKafkaInstrumentor().instrument()` to automatically trace `AIOKafkaProducer` and `AIOKafkaConsumer` operations. Ensure a Kafka broker is running and accessible at `localhost:9092` (or via the `KAFKA_BOOTSTRAP_SERVERS` environment variable).
import asyncio
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
from opentelemetry.instrumentation.aiokafka import AioKafkaInstrumentor
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
# Configure OpenTelemetry TracerProvider
provider = TracerProvider()
processor = SimpleSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
# Instrument aiokafka
AioKafkaInstrumentor().instrument()
KAFKA_BOOTSTRAP_SERVERS = os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
KAFKA_TOPIC = "test_topic"
async def produce_message():
producer = AIOKafkaProducer(bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS)
await producer.start()
try:
with trace.get_tracer(__name__).start_as_current_span("send_test_message"):
message = b"Hello OpenTelemetry from AioKafka!"
print(f"Producing message: {message.decode()}")
await producer.send_and_wait(KAFKA_TOPIC, message)
print("Message produced.")
finally:
await producer.stop()
async def consume_messages():
consumer = AIOKafkaConsumer(
KAFKA_TOPIC,
bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
group_id="my-otel-group",
auto_offset_reset="earliest"
)
await consumer.start()
try:
print("Consumer started, waiting for messages...")
# Consume at least one message
async for msg in consumer:
with trace.get_tracer(__name__).start_as_current_span("process_kafka_message"):
print(f"Consumed: Topic={msg.topic}, Partition={msg.partition}, Offset={msg.offset}, Value={msg.value.decode()}")
break # For quickstart, consume one and exit
finally:
await consumer.stop()
async def main():
# Ensure Kafka is running at KAFKA_BOOTSTRAP_SERVERS before running.
# You might need to create the topic 'test_topic' manually or configure Kafka for auto-creation.
producer_task = asyncio.create_task(produce_message())
consumer_task = asyncio.create_task(consume_messages())
await asyncio.gather(producer_task, consumer_task)
if __name__ == "__main__":
print(f"Using Kafka brokers: {KAFKA_BOOTSTRAP_SERVERS}")
asyncio.run(main())
Debug
Known issues
breakingThis instrumentation library is in beta (`0.x.x` versioning with a `b` suffix), indicating that its API and behavior may change without strict adherence to semantic versioning. Always pin your version and review release notes for potential breaking changes between minor versions.fixPin the exact version (e.g., `opentelemetry-instrumentation-aiokafka==0.62b0`) and carefully review `opentelemetry-python-contrib` changelogs for updates before upgrading.
affects: 0.x.x (all beta versions)
gotchaEnsure the underlying `aiokafka` library version is compatible with the instrumentation. Past issues (e.g., with `0.52b0` and `aiokafka~=0.12.0`) have shown compatibility problems leading to runtime errors, particularly with `AIOKafkaProducer.send()` arguments.fixRefer to the `opentelemetry-python-contrib` GitHub issues and `aiokafka` documentation for known compatible versions. Test thoroughly after any updates to `aiokafka` or the instrumentation.
affects: All versions, especially older beta releases
gotchaThe `instrument()` method must be called early in your application's lifecycle, before `aiokafka` `AIOKafkaProducer` or `AIOKafkaConsumer` instances are created. If `instrument()` is called too late, existing `aiokafka` objects will not be instrumented.fixPlace `AioKafkaInstrumentor().instrument()` at the very beginning of your application's entry point, after configuring the OpenTelemetry SDK.
affects: All versions
gotchaFor the instrumentation to generate visible traces, a Kafka broker must be running and accessible at the specified `bootstrap_servers`. The provided quickstart assumes `localhost:9092`.fixStart a Kafka broker locally (e.g., via Docker or a local installation) or configure `KAFKA_BOOTSTRAP_SERVERS` to point to a running instance.
affects: All versions
gotchaThe quickstart uses a `ConsoleSpanExporter`, which prints traces to standard output. For production environments, you will need to configure a more robust exporter (e.g., OTLP Exporter) to send traces to an observability backend like Jaeger, Zipkin, or an OpenTelemetry Collector.fixReplace `ConsoleSpanExporter` with a suitable exporter, such as `OtlpSpanExporter`, and configure it with the appropriate endpoint and credentials.
affects: All versions
Upgrade
Version history
0.65b0latest on PyPI · released Jul 16, 2026
Audit
Dependencies
aiokafkarequiredPeer dependency, required for actual Kafka communication.
opentelemetry-sdkrequiredCore OpenTelemetry SDK for tracer provider and exporters.
opentelemetry-apirequiredCore OpenTelemetry API for tracing interfaces.