## File: docs/apidoc/KafkaAdminClient.rst

KafkaAdminClient
================

.. autoclass:: kafka.KafkaAdminClient
    :members:

:class:`~kafka.KafkaAdminClient` is composed of functional mixins. The
sections below group its methods by area.


Topics
------

.. automethod:: kafka.KafkaAdminClient.list_topics
.. automethod:: kafka.KafkaAdminClient.describe_topics
.. automethod:: kafka.KafkaAdminClient.create_topics
.. automethod:: kafka.KafkaAdminClient.wait_for_topics
.. automethod:: kafka.KafkaAdminClient.delete_topics


Partitions
----------

.. automethod:: kafka.KafkaAdminClient.create_partitions
.. automethod:: kafka.KafkaAdminClient.delete_records
.. automethod:: kafka.KafkaAdminClient.elect_leaders
.. automethod:: kafka.KafkaAdminClient.alter_partition_reassignments
.. automethod:: kafka.KafkaAdminClient.list_partition_reassignments
.. automethod:: kafka.KafkaAdminClient.describe_topic_partitions
.. automethod:: kafka.KafkaAdminClient.list_partition_offsets


Configs
-------

.. automethod:: kafka.KafkaAdminClient.describe_configs
.. automethod:: kafka.KafkaAdminClient.list_config_resources
.. automethod:: kafka.KafkaAdminClient.alter_configs
.. automethod:: kafka.KafkaAdminClient.reset_configs


Consumer Groups
---------------

.. automethod:: kafka.KafkaAdminClient.list_groups
.. automethod:: kafka.KafkaAdminClient.describe_groups
.. automethod:: kafka.KafkaAdminClient.list_group_offsets
.. automethod:: kafka.KafkaAdminClient.alter_group_offsets
.. automethod:: kafka.KafkaAdminClient.reset_group_offsets
.. automethod:: kafka.KafkaAdminClient.delete_group_offsets
.. automethod:: kafka.KafkaAdminClient.delete_groups
.. automethod:: kafka.KafkaAdminClient.remove_group_members


ACLs
----

.. automethod:: kafka.KafkaAdminClient.describe_acls
.. automethod:: kafka.KafkaAdminClient.create_acls
.. automethod:: kafka.KafkaAdminClient.delete_acls


Cluster
-------

.. automethod:: kafka.KafkaAdminClient.describe_cluster
.. automethod:: kafka.KafkaAdminClient.describe_log_dirs
.. automethod:: kafka.KafkaAdminClient.alter_replica_log_dirs
.. automethod:: kafka.KafkaAdminClient.describe_metadata_quorum
.. automethod:: kafka.KafkaAdminClient.get_broker_version_data
.. automethod:: kafka.KafkaAdminClient.api_versions
.. automethod:: kafka.KafkaAdminClient.describe_features
.. automethod:: kafka.KafkaAdminClient.update_features


Transactions
------------

.. automethod:: kafka.KafkaAdminClient.list_transactions
.. automethod:: kafka.KafkaAdminClient.describe_transactions
.. automethod:: kafka.KafkaAdminClient.describe_producers
.. automethod:: kafka.KafkaAdminClient.abort_transaction
.. automethod:: kafka.KafkaAdminClient.find_hanging_transactions


Users
-----

.. automethod:: kafka.KafkaAdminClient.alter_user_scram_credentials
.. automethod:: kafka.KafkaAdminClient.describe_user_scram_credentials

---

## File: docs/apidoc/modules.rst

kafka-python API
****************

The :mod:`kafka` package exposes a small set of public classes for working
with Apache Kafka. The following sections group them by role; each class
links to its own API reference page.


Application code typically interacts with one of three top-level clients.
Each owns a background IO thread and a shared async networking layer.
The sections below group each client with the data types and extension
points specific to it.


kafka.consumer
==============

The high-level, group-aware message consumer and the data types it
produces.

- :class:`~kafka.KafkaConsumer` - high-level, group-aware message consumer.
  Iterable, with manual or automatic offset commits, cooperative rebalance,
  pluggable deserializers, and transactional-read isolation.
- :class:`~kafka.consumer.fetcher.ConsumerRecord` - a single record
  consumed from a topic partition, as yielded by
  :meth:`~kafka.KafkaConsumer.poll` and consumer iteration.
- :class:`~kafka.ConsumerRebalanceListener` - base class for receiving
  partition join/revoke callbacks during a group rebalance. Also includes
  the async interface :class:`~kafka.AsyncConsumerRebalanceListener`.

.. toctree::
   :maxdepth: 1
   :hidden:

   KafkaConsumer <KafkaConsumer>
   ConsumerRecord <misc/ConsumerRecord>
   ConsumerRebalanceListener <misc/ConsumerRebalanceListener>


kafka.producer
==============

The high-level, asynchronous message producer and the data types and
extension points specific to it.

- :class:`~kafka.KafkaProducer` - high-level, asynchronous message producer.
  Batches records into a background sender thread, with optional
  idempotence, transactions, compression, and pluggable serializers.
- :class:`~kafka.producer.future.FutureRecordMetadata` - asynchronous
  handle returned by :meth:`~kafka.KafkaProducer.send`; resolves to a
  ``RecordMetadata`` once the record is acknowledged.
- :class:`~kafka.producer.future.RecordMetadata` - metadata about a
  produced record after the broker has acknowledged it.
- :class:`~kafka.partitioner.Partitioner` - base class for pluggable
  partition selection; controls which partition a record is assigned to.

.. toctree::
   :maxdepth: 1
   :hidden:

   KafkaProducer <KafkaProducer>
   FutureRecordMetadata <misc/FutureRecordMetadata>
   RecordMetadata <misc/RecordMetadata>
   Partitioner <misc/Partitioner>


kafka.admin
===========

Cluster administration operations.

- :class:`~kafka.KafkaAdminClient` - admin operations: topic, ACL, config,
  consumer group, partition, quota, log-directory, and quorum management.

.. toctree::
   :maxdepth: 1
   :hidden:

   KafkaAdminClient <KafkaAdminClient>


kafka.net
=========

The clients share a single async networking layer (``kafka.net``). These
classes are exposed for advanced use cases - embedding the connection
pool, building a custom client on top of the kafka.net event loop, or
driving the protocol layer directly from the REPL.

- :mod:`~kafka.net.manager` - connection pool and
  high-level facade over the shared IO event loop. Each top-level client
  owns one.
- :mod:`~kafka.net.connection` - per-broker async
  connection: state machine, request/response correlation, and SASL
  handshake.
- :mod:`~kafka.net.backend.transport` - Async socket I/O with write buffering,
  pause/resume hooks, and the asyncio-shaped protocol callback surface.
- :mod:`~kafka.net.http_connect` - Tunnels broker connections through
  an HTTP CONNECT proxy (RFC 7231).
- :mod:`~kafka.net.socks5` - SOCKS5 client with optional username/password
  authentication.

.. toctree::
   :maxdepth: 1
   :hidden:

   manager <net/manager>
   connection <net/connection>
   transport <net/backend/transport>
   http_connect <net/http_connect>
   socks5 <net/socks5>


other / misc
============

Lightweight data types shared across the clients (and useful when
working with the lower-level protocol layer).

- :class:`~kafka.cluster.ClusterMetadata` - in-memory cache of brokers,
  topics, partitions, and the active controller. Refreshes itself on the
  shared IO thread.
- :class:`~kafka.TopicPartition` - namedtuple identifying a partition as
  ``(topic, partition)``.
- :class:`~kafka.OffsetAndMetadata` - committed-offset record
  ``(offset, metadata, leader_epoch)``.
- :class:`~kafka.OffsetSpec` - enum for partition offset queries.
- :class:`~kafka.IsolationLevel` - enum for transactional isolation.
- :class:`~kafka.Serializer` - base class for serialization / deserialization
  of key and value bytes. Includes helper classes :class:`~kafka.DefaultSerializer`
  and :class:`~kafka.JsonSerializer`.

.. toctree::
   :maxdepth: 1
   :hidden:

   ClusterMetadata <misc/ClusterMetadata>
   TopicPartition <misc/TopicPartition>
   OffsetAndMetadata <misc/OffsetAndMetadata>
   OffsetSpec <misc/OffsetSpec>
   IsolationLevel <misc/IsolationLevel>
   Serializer <misc/Serializer>

---

## File: docs/compatibility.rst

Compatibility
-------------

.. image:: https://img.shields.io/badge/kafka-4.3--0.8-brightgreen.svg
    :target: https://kafka-python.readthedocs.io/compatibility.html
.. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
    :target: https://pypi.python.org/pypi/kafka-python

kafka-python is compatible with (and tested against) broker versions 4.3
through 0.8.0 . kafka-python is not compatible with the 0.8.2-beta release.

Because the kafka server protocol is backwards compatible, kafka-python is
expected to work with newer broker releases as well.

Although kafka-python is tested and expected to work on recent broker versions,
not all features are supported. See :doc:`kips` for per-feature coverage of
Kafka Improvement Proposals, or the github open issues for feature tracking.
PRs welcome!

kafka-python is tested on python 3.8-3.14.
python 2.7 was supported through kafka-python release 2.3.
python 3 is required for kafka-python release 3.0.

Builds and tests via Github Actions Workflows.  See https://github.com/dpkp/kafka-python/actions

---

## File: docs/index.rst

kafka-python
############

.. image:: https://img.shields.io/pypi/v/kafka-python.svg
    :target: https://pypi.org/project/kafka-python
.. image:: https://img.shields.io/badge/kafka-4.3--0.8-brightgreen.svg
    :target: https://kafka-python.readthedocs.io/en/master/compatibility.html

.. image:: https://img.shields.io/badge/license-Apache%202-blue.svg
    :target: https://github.com/dpkp/kafka-python/blob/master/LICENSE
.. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
    :target: https://pypi.python.org/pypi/kafka-python


kafka-python is a pure-python client library for Apache Kafka, the distributed
stream processing engine. It has no external dependencies and no Cython/C/rust
core, making installation across a wide variety of environments simple and easy
to manage. It provides high-level class components for consumer, producer, and
admin clients, as well as CLI scripts for quick interactive tasks.

``kafka-python admin`` serves as a simple alternative to the apache kafka bin/
scripts, particularly if/when you do not have easy access to an
installed/compatible jvm. The CLI interface for admin commands is provided as
``kafka-python admin`` and ``python -m kafka.admin``.

Users looking to add more raw throughput can ``pip install crc32c`` as
an optional dependency, offloading one of the most CPU intensive subsystems
to an optimized C library.


.. code-block:: bash

    pip install kafka-python

    # callable as module or as cli-script
    kafka-python admin -b localhost:9092 cluster describe

    # Create a topic with the admin cli
    python -m kafka.admin -b localhost:9092 topics create -t foo-topic

    # Produce messages
    echo "foo message" | python -m kafka.producer -b localhost:9092 -t foo-topic

    # Consume messages
    python -m kafka.consumer -b localhost:9092 -C auto_offset_reset=earliest -C consumer_timeout_ms=1000 -g foo-group -t foo-topic


What's New in 3.0
*****************

- Protocol Stack dynamically generated from Apache Kafka json message schemas.
- Encode/decode performance optimizations with compiled/cached python bytecode.
- Expanded KIP feature support, including Cooperative Rebalance (KIP-429),
  Rack-aware Fetch (KIP-392), Log-Truncation detection (KIP-320), Transactional
  Producer improvements (KIP-360, KIP-447, KIP-654), Sticky Partitioner (KIP-480),
  and splittting oversized producer batches (KIP-126).
- Full refactor and expansion of KafkaAdminClient.
- Networking changes to leverage kafka.net event-loop and async/await syntax.
- Python 3.8+ required

KafkaConsumer
*************

:class:`~kafka.KafkaConsumer` is a high-level message consumer, intended to operate
as similarly as possible to the official java client.
See https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html
for API and configuration details.

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

.. code-block:: python

    from kafka import KafkaConsumer
    consumer = KafkaConsumer('my_favorite_topic')
    for msg in consumer:
        print (msg)

.. code-block:: python

    # join a consumer group for dynamic partition assignment and offset commits
    from kafka import KafkaConsumer
    consumer = KafkaConsumer('my_favorite_topic', group_id='my_favorite_group')
    for msg in consumer:
        print (msg)

.. code-block:: python

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

Keys and Values returned by KafkaConsumer will be raw bytes by default. Use a
``value_deserializer`` to automatically decode into something else. Helpers
are available for simple utf-8 string decoding (``DefaultSerializer``) and
json (``JsonSerializer``).

.. code-block:: python

    # Deserialize json-encoded values
    from kafka import KafkaConsumer, JsonSerializer
    consumer = KafkaConsumer(value_deserializer=JsonSerializer())
    consumer.subscribe(['json-foo'])
    for msg in consumer:
        assert isinstance(msg.value, dict)

.. code-block:: python

    # Access record headers. The returned value is a list of
    # (str, bytes) tuples, representing the header key and value.
    for msg in consumer:
        print (msg.headers)

.. code-block:: python

    # Read only committed messages from transactional topic
    from kafka import KafkaConsumer, IsolationLevel
    consumer = KafkaConsumer(isolation_level=IsolationLevel.READ_COMMITTED)
    consumer.subscribe(['txn_topic'])
    for msg in consumer:
        print(msg)

.. code-block:: python

    # Get consumer metrics
    metrics = consumer.metrics()


KafkaProducer
*************

:class:`~kafka.KafkaProducer` is a high-level, asynchronous message producer.
The class is intended to operate as similarly as possible to the official java
client. See `KafkaProducer <apidoc/KafkaProducer.html>`_ for more details.

.. code-block:: python

    from kafka import KafkaProducer
    producer = KafkaProducer(bootstrap_servers='localhost:1234')
    for _ in range(100):
        # Fire-and-forget: send() is async and returns before delivery
        producer.send('foobar', b'some_message_bytes')

.. code-block:: python

    # To check the status of an async message delivery, use .get()
    future = producer.send('foobar', b'another_message')
    # future.get() will block until it can return the result or raise on error
    result = future.get(timeout=60)

.. code-block:: python

    # 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_ms
    producer.flush()

Message keys are used to hash messages with the same key to the same
partition. Both keys and values should be raw bytes unless a serializer is configured.

.. code-block:: python

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

.. code-block:: python

    # Serialize json messages
    from kafka import KafkaProducer, JsonSerializer
    producer = KafkaProducer(value_serializer=JsonSerializer())
    producer.send('fizzbuzz', {'foo': 'bar'})

.. code-block:: python

    # Serialize string keys
    from kafka import KafkaProducer, DefaultSerializer
    producer = KafkaProducer(key_serializer=DefaultSerializer())
    producer.send('flipflap', key='ping', value=b'1234')

Compression can be used to reduce message size on the wire. Gzip is supported
via python stdlib. For other compression types you must install optional
dependencies.

.. code-block:: python

    # Compress messages
    producer = KafkaProducer(compression_type='gzip')
    for i in range(1000):
        producer.send('foobar', b'msg %d' % i)

KafkaProducer also supports transactions and message headers when needed.

.. code-block:: python

    # Use transactions
    producer = KafkaProducer(transactional_id='fizzbuzz')
    producer.init_transactions()
    producer.begin_transaction()
    future = producer.send('txn_topic', value=b'yes')
    future.get() # wait for successful produce
    producer.commit_transaction() # commit the transaction

    producer.begin_transaction()
    future = producer.send('txn_topic', value=b'no')
    future.get() # wait for successful produce
    producer.abort_transaction() # abort the transaction

.. code-block:: python

    # 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')])

.. code-block:: python

    # Get producer performance metrics
    metrics = producer.metrics()


Module CLI Interface
********************

kafka-python also provides simple command-line interfaces for consumer, producer, and admin clients.
Access via ``python -m kafka.consumer``, ``python -m kafka.producer``, and ``python -m kafka.admin``.
See `CLI <cli/index.html>`_ for the full reference.


Compression
***********

kafka-python supports the following compression formats:

- gzip (via stdlib)
- LZ4 (via `python-lz4`, `lz4tools`, or `py-lz4framed`)
- Snappy (via `python-snappy`)
- Zstandard (via `python-zstandard`)

gzip is supported natively, the others require installing additional libraries.
See `Install `_ for more information.


Optimized CRC32 Validation
**************************

Kafka uses CRC32 checksums to validate messages. kafka-python includes a pure
python implementation for compatibility. To improve performance for high-throughput
applications, kafka-python will use `crc32c` for optimized native code if installed.
See `Install `_ for installation instructions.
See https://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 layer
for interacting with kafka brokers via the python repl. This is useful for
testing, probing, and general experimentation. In version 3.0 the protocol
layer was re-written to generate encoder/decoder classes using json message
definitions imported directly from the Apache Kafka project source.


Debugging
*********

Use python's `logging` module to view internal operational events.
See https://docs.python.org/3/howto/logging.html for overview / howto.

.. code-block:: python

    import logging
    logging.basicConfig(level=logging.DEBUG)


.. toctree::
   :hidden:
   :maxdepth: 2

   High-level Clients <high_level>
   Command Line Interface <cli/index>
   API </apidoc/modules>
   install
   tests
   compatibility
   Supported KIPs <kips>
   support
   license
   Upgrading to 3.0 
   changelog