KafkaConsumer

KafkaConsumer is a high-level message consumer, intended to operate as similarlyas possible to the official java client. Full support for coordinatedconsumer groups requires use of kafka brokers that support the Group APIs: kafka v0.9+.

Seehttps://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.htmlfor API and configuration details.

The consumer iterator returns ConsumerRecords, which are simple namedtuplesthat expose basic message attributes: topic, partition, offset, key, and value:

fromkafkaimportKafkaConsumerconsumer=KafkaConsumer('my_favorite_topic')formsginconsumer:print(msg)

# join a consumer group for dynamic partition assignment and offset commitsfromkafkaimportKafkaConsumerconsumer=KafkaConsumer('my_favorite_topic',group_id='my_favorite_group')formsginconsumer:print(msg)

# manually assign the partition list for the consumerfromkafkaimportTopicPartitionconsumer=KafkaConsumer(bootstrap_servers='localhost:1234')consumer.assign([TopicPartition('foobar',2)])msg=next(consumer)

# Deserialize msgpack-encoded valuesconsumer=KafkaConsumer(value_deserializer=msgpack.loads)consumer.subscribe(['msgpackfoo'])formsginconsumer:assertisinstance(msg.value,dict)

# Access record headers. The returned value is a list of tuples# with str, bytes for key and valueformsginconsumer:print(msg.headers)

# Read only committed messages from transactional topicconsumer=KafkaConsumer(isolation_level='read_committed')consumer.subscribe(['txn_topic'])formsginconsumer:print(msg)

# Get consumer metricsmetrics=consumer.metrics()

KafkaProducer

KafkaProducer is a high-level, asynchronous message producer. The class isintended to operate as similarly as possible to the official java client.Seehttps://kafka-python.readthedocs.io/en/master/apidoc/KafkaProducer.htmlfor more details.

fromkafkaimportKafkaProducerproducer=KafkaProducer(bootstrap_servers='localhost:1234')for_inrange(100):producer.send('foobar',b'some_message_bytes')

# Block until a single message is sent (or timeout)future=producer.send('foobar',b'another_message')result=future.get(timeout=60)

# Block until all pending messages are at least put on the network# NOTE: This does not guarantee delivery or success! It is really# only useful if you configure internal batching using linger_msproducer.flush()

# Use a key for hashed-partitioningproducer.send('foobar',key=b'foo',value=b'bar')

# Serialize json messagesimportjsonproducer=KafkaProducer(value_serializer=lambdav:json.dumps(v).encode('utf-8'))producer.send('fizzbuzz',{'foo':'bar'})

# Serialize string keysproducer=KafkaProducer(key_serializer=str.encode)producer.send('flipflap',key='ping',value=b'1234')

# Compress messagesproducer=KafkaProducer(compression_type='gzip')foriinrange(1000):producer.send('foobar',b'msg%d'%i)

# Use transactionsproducer=KafkaProducer(transactional_id='fizzbuzz')producer.init_transactions()producer.begin_transaction()future=producer.send('txn_topic',value=b'yes')future.get()# wait for successful produceproducer.commit_transaction()# commit the transactionproducer.begin_transaction()future=producer.send('txn_topic',value=b'no')future.get()# wait for successful produceproducer.abort_transaction()# abort the transaction

# Include record headers. The format is list of tuples with string key# and bytes value.producer.send('foobar',value=b'c29tZSB2YWx1ZQ==',headers=[('content-encoding',b'base64')])

# Get producer performance metricsmetrics=producer.metrics()

Thread safety

The KafkaProducer can be used across threads without issue, unlike theKafkaConsumer which cannot.

While it is possible to use the KafkaConsumer in a thread-local manner,multiprocessing is recommended.

Compression

kafka-python supports the following compression formats:

• gzip

• LZ4

• Snappy

• Zstandard (zstd)

gzip is supported natively, the others require installing additional libraries.Seehttps://kafka-python.readthedocs.io/en/master/install.html for more information.

Optimized CRC32 Validation

Kafka uses CRC32 checksums to validate messages. kafka-python includes a purepython implementation for compatibility. To improve performance for high-throughputapplications, kafka-python will usecrc32c for optimized native code if installed.Seehttps://kafka-python.readthedocs.io/en/master/install.html for installation instructions.Seehttps://pypi.org/project/crc32c/ for details on the underlying crc32c lib.

Protocol

A secondary goal of kafka-python is to provide an easy-to-use protocol layerfor interacting with kafka brokers via the python repl. This is useful fortesting, probing, and general experimentation. The protocol support isleveraged to enable a KafkaClient.check_version() method thatprobes a kafka broker and attempts to identify which version it is running(0.8.0 to 2.6+).