PyPIDataupdated 08 Aug 2026

kafka-python

Item: kafka-python
Rating: 4.3
Author: Keyoor

kafka-python is a pure-Python client for Apache Kafka brokers from 0.8 through 4.3. It provides synchronous high-level producers and consumers, a broad admin client, lower-level protocol access, and command-line tools that can inspect clusters, create topics, publish records, and consume without Kafka's JVM scripts. Version 3 rewrote the protocol layer from Apache Kafka's JSON message schemas, expanded modern KIP support, and made producer idempotence plus all-replica acknowledgments the defaults. Installation is simple because the core has no compiled extension, but operating a correct Kafka consumer or transactional producer still requires careful broker, offset, timeout, and security configuration.

kafka-python · v3.0.10 · 6.7M/wk · ★ 5,901 · pushed 07 Aug 2026 · dpkp/kafka-python · docs · reviewed by Keyoor

Verdict

kafka-python is the best all-Python choice when portability and inspectable code matter more than native-client throughput. Treat its simple install as only the beginning: delivery, offsets, rebalances, security, and transactions still demand explicit production configuration.

API stability3/5Producer and consumer concepts track the official Java client closely, and familiar classes remain in 3.0. Version 3 nevertheless rewrote the protocol stack, refactored and expanded KafkaAdminClient, changed internal networking, and adopted idempotence plus acks='all' defaults. The admin docs now deprecate lists of NewTopic objects in favor of names or option dictionaries. This is healthy modernization, but a major upgrade deserves integration tests against the exact broker versions.

Docs5/5The README gives runnable producer, consumer, transaction, header, compression, metrics, CLI, and serializer examples, while the API reference documents defaults and broker-version constraints in detail. It explains that send is asynchronous, flush is not delivery confirmation, idempotence can be disabled by conflicting values, consumer auto-commit is periodic, iteration blocks forever by default, and clients are not thread-safe. That operational candor is excellent.

Maintenance5/5PyPI lists 3.0.10, GitHub was pushed on 2026-08-07, and the repository shows only 18 open issues and pull requests with 5,901 stars. Version 3 adds protocol definitions derived from Apache Kafka, expanded KIP coverage, modern producer semantics, and a substantially expanded admin client. The repository is not archived and currently advertises compatibility through Kafka 4.3, strong evidence of active protocol maintenance.

Ecosystem4/5The package recorded millions of weekly downloads, spans producers, consumers, admin operations, CLI use, serializers, transactions, metrics, and low-level protocol experiments, and needs no core dependency beyond Python. Optional compression and crc32c packages cover common performance needs. It lacks the native speed, Schema Registry path, and commercial ecosystem around confluent-kafka, and async applications often choose aiokafka instead.

Use it if

You want a pure-Python Kafka client that installs without a C, Cython, Rust, or JVM runtime
You need producer, consumer, admin, CLI, and low-level protocol access from one package
Your throughput fits Python and you value readable internals over maximum native-client performance
You need compatibility across a wide Kafka broker range, including modern idempotent and transactional features

Skip it if

You need the highest throughput or lowest CPU use: the README calls out optional native crc32c specifically because checksum work is otherwise pure Python
Your application is asyncio-first and needs awaitable public consumer and producer APIs: the high-level examples remain synchronous even though v3 refactored internal networking
You want vendor support, Schema Registry integration, and close parity with librdkafka configuration: confluent-kafka is the better fit
You expect flush to prove delivery: the README explicitly says flush only puts pending messages on the network and does not guarantee delivery or success
You need consumers shared across threads: the API documentation says KafkaConsumer is not thread-safe

Setup reality

Core installation is unusually easy: kafka-python 3.0.10 requires Python 3.8 or newer and declares no mandatory runtime dependencies. Optional extras add crc32c acceleration and LZ4, Snappy, or Zstandard compression; gzip works through the standard library. That is where easy ends. bootstrap_servers only discovers the cluster, so provide more than one broker address for startup resilience. The default security protocol is PLAINTEXT. Managed or production clusters usually require SASL_SSL or SSL, a CA file, mechanism, username, password or token provider, and broker hostnames that pass certificate verification. Keep secrets in a secret store rather than source. Producer.send is asynchronous and returns a future. In v3, acks='all', infinite retries, and idempotence are defaults, but explicitly setting acks to 0 or 1, retries=0, or more than five in-flight requests silently disables default-driven idempotence with a warning. Set enable_idempotence=True explicitly when the guarantee is required so conflicts become configuration errors, and wait on futures or attach callbacks because flush alone does not confirm success. Consumers return raw bytes unless deserializers are configured. group_id defaults to None, auto_offset_reset defaults to latest, and iteration blocks forever unless consumer_timeout_ms is set. With a group, enable_auto_commit defaults to True every 5 seconds, which can commit work before your application finishes. Disable it and commit only after successful side effects when at-least-once processing matters. Processing must return to poll within max_poll_interval_ms or the group rebalances. subscribe and assign are mutually exclusive; assign also bypasses group partition management. Transactions require a unique stable transactional_id per producer instance, durable broker topic settings, and consumers configured with read_committed. Finally, max message, fetch, memory, batch, linger, delivery timeout, session, and poll settings must agree with broker limits and real processing latency. Instrument producer and consumer metrics and enable Python logging before guessing at coordinator or network failures.

Patterns

Send JSON and wait for broker acknowledgmentproduce-json

from kafka import JsonSerializer, KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=['kafka-1:9092', 'kafka-2:9092'],
    value_serializer=JsonSerializer(),
    enable_idempotence=True,
    client_id='orders-api',
)
metadata = producer.send('orders', {'id': 42, 'status': 'created'}).get(timeout=30)
producer.close()

send returns immediately. Calling get surfaces delivery errors; explicit idempotence makes conflicting producer settings fail instead of silently weakening the guarantee.

Keep one entity's records on a partitionpartition-by-key

from kafka import DefaultSerializer, JsonSerializer, KafkaProducer

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    key_serializer=DefaultSerializer(),
    value_serializer=JsonSerializer(),
)
producer.send('account-events', key='account-42', value={'type': 'updated'}).get(timeout=30)

Keys influence partition selection and ordering within that partition. A key does not create ordering across the whole topic.

Trade a little latency for compressed batchesbatch-and-compress

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    compression_type='gzip',
    linger_ms=10,
    batch_size=64 * 1024,
    delivery_timeout_ms=120000,
)

gzip needs no extra package. LZ4, Snappy, and Zstandard require optional dependencies, and larger batches consume memory per active partition.

Consume JSON in a managed groupconsume-json-group

from kafka import JsonSerializer, KafkaConsumer

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['kafka-1:9092', 'kafka-2:9092'],
    group_id='billing-v1',
    value_deserializer=JsonSerializer(),
    auto_offset_reset='earliest',
    client_id='billing-worker',
)
for record in consumer:
    handle(record.value)

The iterator blocks forever by default. group_id enables coordinated partition assignment and offset commits.

Commit only after side effects succeedcommit-after-processing

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    group_id='billing-v1',
    enable_auto_commit=False,
    auto_offset_reset='earliest',
    value_deserializer=JsonSerializer(),
)
for record in consumer:
    process_idempotently(record.value)
    consumer.commit()

Manual commits reduce commit-before-work risk but do not create exactly-once external side effects. Processing should itself be idempotent.

Stop a fixture consumer when idlebound-test-consumer

consumer = KafkaConsumer(
    'events',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    consumer_timeout_ms=3000,
)
records = list(consumer)

Without consumer_timeout_ms, iteration blocks forever. This timeout ends after inactivity and is useful for tests, not a continuous worker.

Hide aborted transactional recordsread-committed

from kafka import IsolationLevel, KafkaConsumer

consumer = KafkaConsumer(
    'payments',
    bootstrap_servers='localhost:9092',
    group_id='ledger-v1',
    isolation_level=IsolationLevel.READ_COMMITTED,
)

read_committed is required for end-to-end transaction visibility. The producer and broker topics must also be configured for transactions.

Write several records atomicallyproduce-transaction

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    transactional_id='checkout-shard-3',
    value_serializer=JsonSerializer(),
)
producer.init_transactions()
producer.begin_transaction()
try:
    producer.send('orders', {'id': 42}).get(timeout=30)
    producer.send('outbox', {'order_id': 42}).get(timeout=30)
    producer.commit_transaction()
except Exception:
    producer.abort_transaction()
    raise

transactional_id must be stable and unique to one producer instance or fencing occurs. Consumers need read_committed to hide aborted writes.

Connect to a SCRAM-protected clusterconfigure-sasl-tls

import os
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=os.environ['KAFKA_BOOTSTRAP'].split(','),
    security_protocol='SASL_SSL',
    sasl_mechanism='SCRAM-SHA-512',
    sasl_plain_username=os.environ['KAFKA_USERNAME'],
    sasl_plain_password=os.environ['KAFKA_PASSWORD'],
    ssl_cafile='/etc/ssl/certs/cluster-ca.pem',
)

Hostname verification defaults on. Do not disable it to work around a certificate whose names do not match the brokers.

Read a partition without group assignmentassign-partition

from kafka import KafkaConsumer, TopicPartition

consumer = KafkaConsumer(bootstrap_servers='localhost:9092')
partition = TopicPartition('audit', 2)
consumer.assign([partition])
consumer.seek(partition, 100)
record = next(consumer)

assign is incompatible with subscribe and bypasses group rebalancing. Offsets are not committed unless a group_id and explicit commit strategy are used.

Create a topic and wait for metadatacreate-topic

from kafka import KafkaAdminClient

admin = KafkaAdminClient(bootstrap_servers='localhost:9092')
admin.create_topics(
    {
        'orders': {
            'num_partitions': 6,
            'replication_factor': 3,
            'configs': {'min.insync.replicas': '2'},
        }
    },
    wait_for_metadata=True,
)

Version 3 deprecates lists of NewTopic objects. Replication factor and broker count must agree, and creation needs cluster authorization.

Inspect network and coordinator activityenable-client-debugging

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger('kafka').setLevel(logging.DEBUG)

Debug logs can be noisy and may reveal broker addresses or metadata. Enable them selectively and protect production log access.

Alternatives

Package	Registry	Pick it when
confluent-kafka	PyPI	You want librdkafka performance, Confluent integrations, and a vendor-backed Python client
aiokafka	PyPI	You need native asyncio producer and consumer APIs in an async service
quixstreams	PyPI	You want a higher-level Python stream-processing framework with state, windows, and dataframe-like operations