kafka-python Documentation

This project tracks support for Kafka Improvement Proposals (KIPs), which define protocol and feature changes in Apache Kafka. Support status in kafka-python refers specifically to whether a feature is usable through the high-level client APIs (KafkaProducer, KafkaConsumer, or KafkaAdminClient), rather than just the existence of the wire format.

Note that as a client library, broker-internal protocols (such as KRaft or inter-broker replication) are intentionally out of scope.

Status Legend:

• Supported: The feature is fully usable through the kafka-python client or admin API.
• Partial: The feature is partially implemented (refer to specific KIP notes for details).
• Protocol only: Wire/protocol classes exist under kafka/protocol/, but the client does not yet drive or expose the feature.
• --: Not implemented or not supported.

To install the latest bleed-edge code directly from the GitHub repository, use the following command.

pip install git+https://github.com/dpkp/kafka-python.git

To enable LZ4 compression and decompression support, install the lz4 extra which includes python-lz4.

pip install 'kafka-python[lz4]'

Use KafkaProducer to send messages to Kafka. The send() method is asynchronous by default and returns a future. You can block for a synchronous send by calling .get() on the future, or use callbacks for asynchronous handling.

from kafka import KafkaProducer, JsonSerializer, DefaultSerializer
from kafka.errors import KafkaError

producer = KafkaProducer(bootstrap_servers=['broker1:1234'])

# Asynchronous send
future = producer.send('my-topic', b'raw_bytes')

# Block for 'synchronous' send
try:
    record_metadata = future.get(timeout=10)
    print (record_metadata.topic)
    print (record_metadata.partition)
    print (record_metadata.offset)
except KafkaError:
    log.exception()

# Produce keyed messages (enables hashed partitioning)
producer.send('keyed-topic', key=b'foo', value=b'bar')

# Block until all async messages are sent
producer.flush()

You can pass configuration parameters to the consumer using the -C flag.

Common configurations include:

• auto_offset_reset: Determines the starting offset (e.g., earliest).
• consumer_timeout_ms: The amount of time to wait for new messages before exiting (in milliseconds).

# Read from the beginning, then exit after 1s of idle
kafka-python consumer -b localhost:9092 -t my-topic \
    -C auto_offset_reset=earliest \
    -C consumer_timeout_ms=1000

By default, kafka-python calculates record checksums in pure Python, which can become a CPU bottleneck as throughput increases. Installing the crc32c dependency uses an optimized C library to reduce CPU cost. This is highly recommended for performance.

pip install 'kafka-python[crc32c]'

To run the unit test suite, which includes mocked network interfaces and broker simulations, first install the development dependencies and then execute pytest or use the provided make command.

Prerequisites:

• A Python environment (preferably a virtualenv).
• requirements-dev.txt must be present in the repository.

The kafka.sasl module has moved to kafka.net.sasl. If you implement a custom AbstractTokenProvider or SASL mechanism, update your imports to use kafka.net.sasl.oauth.

from kafka.net.sasl.oauth import AbstractTokenProvider

You can run the kafka-python CLI tools either as a console script or as Python module entry points. This is useful for running consumer, producer, or admin tasks without needing a JVM-based Apache Kafka installation.

Console script usage: kafka-python <command> [options]

Module invocation usage: python -m kafka.<command> [options]

# Using the console script
kafka-python consumer -b localhost:9092 -t my-topic
kafka-python producer -b localhost:9092 -t my-topic
kafka-python admin    -b localhost:9092 cluster describe

# Using module entry points
python -m kafka.consumer -b localhost:9092 -t my-topic
python -m kafka.producer -b localhost:9092 -t my-topic
python -m kafka.admin    -b localhost:9092 cluster describe

To enable ZSTD compression and decompression support, install the zstd extra which includes python-zstandard.

pip install 'kafka-python[zstd]'

Install the standard version of kafka-python from PyPI using pip.

pip install kafka-python

The kafka-python admin command-line tool exposes KafkaAdminClient operations. Commands are organized by resource type (e.g., topics, partitions, configs).

Output Formats:

• --format raw (default): Uses pprint for human-readable output.
• --format json: Emits a single JSON document, ideal for piping to tools like jq.

Common usage examples:

• Describe the cluster: kafka-python admin -b localhost:9092 cluster describe
• List topics in JSON format: kafka-python admin -b localhost:9092 --format json topics list
• Create a topic via the python module: python -m kafka.admin -b localhost:9092 topics create -t foo --num-partitions 3