### Install Snappy Python Extension

Source: https://github.com/dpkp/kafka-python/wiki/Home

Install the python-snappy library after building it from source.

```bash
python setup.py install
```

--------------------------------

### Install kafka-python from Source with Setuptools

Source: https://github.com/dpkp/kafka-python/wiki/Home

Clone the repository and install kafka-python using the easy_install command.

```bash
git clone https://github.com/mumrah/kafka-python
easy_install ./kafka-python
```

--------------------------------

### Install kafka-python using setup.py

Source: https://github.com/dpkp/kafka-python/wiki/Home

Clone the repository and install kafka-python by directly running the setup.py script.

```bash
git clone https://github.com/mumrah/kafka-python
cd kafka-python
python setup.py install
```

--------------------------------

### Install optional dependencies

Source: https://github.com/dpkp/kafka-python/blob/master/docs/install.rst

Commands to install optional features for performance optimization and compression support.

```bash
pip install 'kafka-python[crc32c]'
```

```pycon
>>> pip install 'kafka-python[zstd]'
```

```pycon
>>> pip install 'kafka-python[lz4]'
```

```bash
pip install 'kafka-python[snappy]'
```

--------------------------------

### Install python-snappy from PyPI

Source: https://github.com/dpkp/kafka-python/wiki/Home

Install the python-snappy library directly from the Python Package Index using pip.

```bash
pip install python-snappy
```

--------------------------------

### Install Snappy Development Libraries (APT)

Source: https://github.com/dpkp/kafka-python/wiki/Home

Install the Snappy C library development files on Debian/Ubuntu systems using apt-get.

```bash
sudo apt-get install libsnappy-dev
```

--------------------------------

### Install latest development version

Source: https://github.com/dpkp/kafka-python/blob/master/docs/install.rst

Install the latest code directly from the GitHub repository.

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

--------------------------------

### Install test dependencies

Source: https://github.com/dpkp/kafka-python/blob/master/docs/tests.rst

Install the required development dependencies before running the test suite.

```bash
pip install -r requirements-dev.txt
```

--------------------------------

### Install and Use kafka-python CLI

Source: https://github.com/dpkp/kafka-python/blob/master/docs/index.rst

Commands for installing the library and performing basic Kafka operations like topic management, producing, and consuming messages via the command line.

```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
```

--------------------------------

### Install kafka-python from Source with Pip

Source: https://github.com/dpkp/kafka-python/wiki/Home

Clone the repository and install the bleeding-edge version of kafka-python using pip.

```bash
git clone https://github.com/mumrah/kafka-python
pip install ./kafka-python
```

--------------------------------

### Install Snappy Development Libraries (YUM)

Source: https://github.com/dpkp/kafka-python/wiki/Home

Install the Snappy C library development files on RPM-based systems using yum.

```bash
sudo yum install libsnappy-devel
```

--------------------------------

### Basic CLI Usage Examples

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Common commands for describing clusters, listing topics in JSON format, and creating topics.

```bash
kafka-python admin -b localhost:9092 cluster describe
kafka-python admin -b localhost:9092 --format json topics list
python -m kafka.admin -b localhost:9092 topics create -t foo --num-partitions 3
```

--------------------------------

### Install Snappy Development Libraries (Brew)

Source: https://github.com/dpkp/kafka-python/wiki/Home

Install the Snappy C library development files on macOS using Homebrew.

```bash
brew install snappy
```

--------------------------------

### Kafka API Version Configuration Examples

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaProducer.rst

Examples of tuple values used to specify the Kafka broker API version for compatibility.

```default
(4, 3) most recent broker release, enable all supported features
(0, 11) enables message format v2 (internal)
(0, 10, 0) enables sasl authentication and message format v1
(0, 9) enables full group coordination features with automatic
    partition assignment and rebalancing,
(0, 8, 2) enables kafka-storage offset commits with manual
    partition assignment only,
(0, 8, 1) enables zookeeper-storage offset commits with manual
    partition assignment only,
(0, 8, 0) enables basic functionality but requires manual
    partition assignment and offset management.
```

--------------------------------

### Common kafka-python consumer CLI commands

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/consumer.rst

Examples for reading records, joining consumer groups, and configuring offset behavior.

```bash
# Read every record on a topic until interrupted
kafka-python consumer -b localhost:9092 -t my-topic

# Join a consumer group and commit offsets automatically
kafka-python consumer -b localhost:9092 -g my-group -t my-topic

# 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

# Print full ConsumerRecord (topic, partition, offset, key, value, ...)
kafka-python consumer -b localhost:9092 -t my-topic -f full
```

--------------------------------

### Install kafka-python with Pip

Source: https://github.com/dpkp/kafka-python/wiki/Home

Use this command to install the latest stable release of kafka-python using pip.

```bash
pip install kafka-python
```

--------------------------------

### Initialize and Use KafkaConsumer

Source: https://github.com/dpkp/kafka-python/blob/master/docs/high_level.rst

Basic setup for consuming messages from a Kafka topic with auto-commit enabled.

```python
from kafka import KafkaConsumer, OffsetAndMetadata, TopicPartition, JsonSerializer, DefaultSerializer

# To consume latest messages and auto-commit offsets
consumer = KafkaConsumer('my-topic',
                         group_id='my-group',
                         bootstrap_servers=['localhost:9092'])
for message in consumer:
    # message value and key are raw bytes -- decode if necessary!
    # e.g., for unicode: `message.value.decode('utf-8')`
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                          message.offset, message.key,
                                          message.value))
```

--------------------------------

### begin_transaction()

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaProducer.rst

Starts a new transaction. Must be called after init_transactions().

```APIDOC
## begin_transaction()

### Description
Should be called before the start of each new transaction. Note that prior to the first invocation of this method, you must invoke init_transactions() exactly one time.

### Raises
- **ProducerFencedError** - If another producer with the same transactional_id is active.
```

--------------------------------

### Get API Versions

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to retrieve supported API versions.

```default
kafka-python admin cluster api-versions [-h] [-k API_KEYS] [--raw]
```

--------------------------------

### Get Broker Version

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to retrieve the version of a specific broker.

```default
kafka-python admin cluster broker-version [-h] --broker BROKER
```

--------------------------------

### Get Producer Performance Metrics

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Shows how to retrieve performance metrics from the producer instance using the `metrics()` method.

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

--------------------------------

### Administer Kafka Cluster with CLI

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Use the kafka-python admin CLI for cluster operations. This example shows how to describe the cluster.

```bash
kafka-python admin -b localhost:9092 cluster describe
```

--------------------------------

### Enable System Site Packages in venv

Source: https://github.com/dpkp/kafka-python/wiki/kafka-import-error

Modify the 'venv/pyvenv.cfg' file to include system site packages. This is necessary when a package installed globally or in another environment is required within the virtual environment, such as when integrating Flask with Kafka.

```text
include-system-site-packages = true
```

--------------------------------

### Create Partitions

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to create additional partitions for existing topics.

```default
kafka-python admin partitions create [-h] -p TOPIC_PARTITIONS
                                     [--timeout-ms TIMEOUT_MS]
                                     [--validate-only]
```

--------------------------------

### kafka-python consumer usage help

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/consumer.rst

Displays the command-line interface usage and available arguments for the consumer tool.

```default
usage: kafka-python consumer [-h] -b BOOTSTRAP_SERVERS [-S SECURITY_PROTOCOL]
                             [-M SASL_MECHANISM] [-U SASL_USER]
                             [-P SASL_PASSWORD] [-l LOG_LEVEL]
                             [-L ENABLE_LOGGER] [-D DISABLE_LOGGER]
                             [--log-format LOG_FORMAT]
                             [--log-date-format LOG_DATE_FORMAT]
                             [--log-file LOG_FILE] [--log-config LOG_CONFIG]
                             [-C EXTRA_CONFIG] -t TOPICS [-g GROUP]
                             [-i GROUP_INSTANCE_ID] [-f FORMAT]
                             [--encoding ENCODING]
```

--------------------------------

### Global CLI Options

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Displays the usage help and available global flags for the admin tool.

```default
usage: kafka-python admin [-h] [-b BOOTSTRAP_SERVERS] [-S SECURITY_PROTOCOL]
                          [-M SASL_MECHANISM] [-U SASL_USER]
                          [-P SASL_PASSWORD] [-l LOG_LEVEL] [-L ENABLE_LOGGER]
                          [-D DISABLE_LOGGER] [--log-format LOG_FORMAT]
                          [--log-date-format LOG_DATE_FORMAT]
                          [--log-file LOG_FILE] [--log-config LOG_CONFIG]
                          [-C EXTRA_CONFIG] [--format FORMAT]
                          GROUP ...
```

--------------------------------

### subscription()

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaConsumer.rst

Get the current topic subscription.

```APIDOC
## subscription()

### Description
Get the current topic subscription.

### Returns
- **set** - The set of subscribed topics.
```

--------------------------------

### topics(timeout_ms=None)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaConsumer.rst

Get all topics the user is authorized to view.

```APIDOC
## topics(timeout_ms=None)

### Description
Get all topics the user is authorized to view. This will always issue a remote call to the cluster.

### Parameters
- **timeout_ms** (numeric) - Optional - Maximum time in milliseconds to block.

### Returns
- **set** - Authorized topics.
```

--------------------------------

### position(partition, timeout_ms=None)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaConsumer.rst

Get the offset of the next record that will be fetched.

```APIDOC
## position(partition, timeout_ms=None)

### Description
Get the offset of the next record that will be fetched.

### Parameters
- **partition** (TopicPartition) - Required - Partition to check
- **timeout_ms** (None) - Optional - Timeout in milliseconds

### Returns
- **int** - Offset or None
```

--------------------------------

### Initialize KafkaProducer and Send Messages

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Demonstrates initializing KafkaProducer and sending messages in a fire-and-forget manner. The `send()` method is asynchronous and returns before message delivery is confirmed.

```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')
```

--------------------------------

### Invoke kafka-python CLI tools

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/index.rst

Demonstrates basic usage of consumer, producer, and admin commands using both the console script and module entry points.

```bash
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

# equivalent module invocations
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
```

--------------------------------

### kafka-python producer usage reference

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/producer.rst

Displays the command-line arguments and options available for the producer.

```default
usage: kafka-python producer [-h] -b BOOTSTRAP_SERVERS [-S SECURITY_PROTOCOL]
                             [-M SASL_MECHANISM] [-U SASL_USER]
                             [-P SASL_PASSWORD] [-l LOG_LEVEL]
                             [-L ENABLE_LOGGER] [-D DISABLE_LOGGER]
                             [--log-format LOG_FORMAT]
                             [--log-date-format LOG_DATE_FORMAT]
                             [--log-file LOG_FILE] [--log-config LOG_CONFIG]
                             [-C EXTRA_CONFIG] -t TOPIC [--encoding ENCODING]
```

--------------------------------

### get(timeout=None)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/misc/FutureRecordMetadata.rst

Blocks the calling thread until the record is acknowledged by the broker or the specified timeout is reached.

```APIDOC
## get(timeout=None)

### Description
Wait for up to timeout seconds for the future to complete. This method blocks until the record is acknowledged or the timeout expires.

### Parameters
- **timeout** (float/None) - Optional - The maximum time to wait in seconds. If None, blocks indefinitely.
```

--------------------------------

### beginning_offsets(partitions, timeout_ms=None)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaConsumer.rst

Fetches the earliest available offsets for the specified partitions.

```APIDOC
## beginning_offsets(partitions, timeout_ms=None)

### Description
Get the first offset for the given partitions. This method does not change the current consumer position.

### Parameters
- **partitions** (list) - Required - List of TopicPartition instances to fetch offsets for.
- **timeout_ms** (int) - Optional - Milliseconds to block fetching offsets. Defaults to default_api_timeout_ms.

### Returns
- **{TopicPartition: int}** - The earliest available offsets for the given partitions.
```

--------------------------------

### Enable Sticky Partitioner

Source: https://github.com/dpkp/kafka-python/blob/master/docs/upgrade_to_3_0.rst

Configure the KafkaProducer to use the StickyPartitioner for improved batching behavior with null keys.

```default
from kafka.partitioner import StickyPartitioner

KafkaProducer(
    bootstrap_servers=...,
    partitioner=StickyPartitioner(),
)
```

--------------------------------

### Get Kafka Consumer Metrics

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Retrieve metrics from a KafkaConsumer instance. This can be useful for monitoring and performance analysis.

```python
metrics = consumer.metrics()
```

--------------------------------

### Create Kafka Topic with Admin CLI

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Create a new topic in Kafka using the admin CLI. Specify the bootstrap server and topic name.

```bash
python -m kafka.admin -b localhost:9092 topics create -t foo-topic
```

--------------------------------

### Describe Partitions

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to describe topic partitions with pagination support for brokers 3.9 and later.

```default
kafka-python admin partitions describe [-h] -t TOPICS
                                       [--response-partition-limit RESPONSE_PARTITION_LIMIT]
                                       [--cursor-topic CURSOR_TOPIC]
                                       [--cursor-partition CURSOR_PARTITION]
```

--------------------------------

### Clone and Navigate to Snappy Repository

Source: https://github.com/dpkp/kafka-python/wiki/Home

Clone the python-snappy library from its GitHub repository and navigate into the directory.

```bash
git clone https://github.com/andrix/python-snappy
cd python-snappy
```

--------------------------------

### Describe Kafka Configs

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to describe existing Kafka configurations.

```default
kafka-python admin configs describe [-h] -r RESOURCE_TYPE -n RESOURCE_NAMES
                                    [-c CONFIGS] [--dynamic] [--modified]
                                    [--static] [--default]
```

--------------------------------

### bootstrap(timeout_ms=None, refresh=False)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaConsumer.rst

Blocks until the consumer has successfully bootstrapped cluster metadata.

```APIDOC
## bootstrap(timeout_ms=None, refresh=False)

### Description
Block until the consumer has bootstrapped cluster metadata. Useful for ensuring metadata is available before issuing the first request.

### Parameters
- **timeout_ms** (int) - Optional - Maximum time to wait in milliseconds. Defaults to None (wait indefinitely).
- **refresh** (bool) - Optional - If True, force a new bootstrap even if one succeeded. Defaults to False.
```

--------------------------------

### Import KafkaProducer in Python

Source: https://github.com/dpkp/kafka-python/wiki/kafka-import-error

This import statement is used to access the KafkaProducer class from the kafka-python library. Ensure the 'kafka' module is correctly installed and accessible within your Python environment.

```python
from kafka import KafkaProducer
```

--------------------------------

### Build Snappy Python Extension

Source: https://github.com/dpkp/kafka-python/wiki/Home

Build the C extension for the python-snappy library after downloading or cloning the source.

```bash
python setup.py build
```

--------------------------------

### Check Asynchronous Message Delivery Status

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Shows how to check the status of an asynchronous message delivery using the `get()` method on the future object returned by `send()`. This call will block until the result is available or an error occurs.

```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)
```

--------------------------------

### resume(*partitions)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaConsumer.rst

Resume fetching from the specified (paused) partitions.

```APIDOC
## resume(*partitions)

### Description
Resume fetching from the specified (paused) partitions.

### Parameters
- **partitions** (TopicPartition) - Required - Partitions to resume.
```

--------------------------------

### Create Kafka Topic

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to create a new Kafka topic with specified partitions and replication factor.

```default
kafka-python admin topics create [-h] -t TOPIC
                                 [--num-partitions NUM_PARTITIONS]
                                 [--replication-factor REPLICATION_FACTOR]
```

--------------------------------

### kafka.KafkaProducer

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaProducer.rst

Initializes a new KafkaProducer instance with the specified configuration parameters.

```APIDOC
## class kafka.KafkaProducer(**configs)

### Description
Initializes a Kafka producer client. This class manages the connection to Kafka brokers and handles message production.

### Parameters
- **allow_auto_create_topics** (bool) - Optional - Enable/disable auto topic creation on metadata request. Default: True.
- **metadata_max_age_ms** (int) - Optional - Time in milliseconds to force metadata refresh. Default: 300000.
- **retry_backoff_ms** (int) - Optional - Milliseconds to backoff when retrying on errors. Default: 100.
- **request_timeout_ms** (int) - Optional - Client request timeout in milliseconds. Default: 30000.
- **default_api_timeout_ms** (int) - Optional - Default timeout for blocking client APIs. Default: 60000.
- **receive_message_max_bytes** (int) - Optional - Maximum allowed network frame size. Default: 100_000_000.
- **receive_buffer_bytes** (int) - Optional - Size of the TCP receive buffer. Default: None.
- **send_buffer_bytes** (int) - Optional - Size of the TCP send buffer. Default: None.
- **socket_options** (list) - Optional - List of tuple-arguments to socket.setsockopt. Default: [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)].
- **reconnect_backoff_ms** (int) - Optional - Time in milliseconds to wait before attempting to reconnect. Default: 50.
- **reconnect_backoff_max_ms** (int) - Optional - Maximum backoff time in milliseconds for reconnections. Default: 30000.
- **socket_connection_setup_timeout_ms** (int) - Optional - Max time to wait for socket connection. Default: 10000.
- **socket_connection_setup_timeout_max_ms** (int) - Optional - Max backoff time for socket connection setup. Default: 30000.
- **max_in_flight_requests_per_connection** (int) - Optional - Maximum requests per broker connection. Default: 5.
- **security_protocol** (str) - Optional - Protocol used to communicate with brokers (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL). Default: PLAINTEXT.
- **ssl_context** (ssl.SSLContext) - Optional - Pre-configured SSLContext. Default: None.
- **ssl_check_hostname** (bool) - Optional - Verify certificate matches broker hostname. Default: True.
- **ssl_cafile** (str) - Optional - Filename of CA file for verification. Default: None.
- **ssl_certfile** (str) - Optional - Filename of client certificate. Default: None.
- **ssl_keyfile** (str) - Optional - Filename of client private key. Default: None.
- **ssl_password** (str) - Optional - Password for certificate chain. Default: None.
- **ssl_crlfile** (str) - Optional - Filename containing CRL. Default: None.
- **ssl_ciphers** (str) - Optional - Available ciphers for SSL.
```

--------------------------------

### Describe Log Directories

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to retrieve log directory statistics for specific brokers or topics.

```default
kafka-python admin cluster describe-log-dirs [-h] [--broker BROKERS]
                                             [--topic TOPICS]
```

--------------------------------

### List Kafka Topics

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to list all available Kafka topics.

```default
kafka-python admin topics list [-h]
```

--------------------------------

### Download and Extract Snappy Source

Source: https://github.com/dpkp/kafka-python/wiki/Home

Download the python-snappy library source archive and extract it.

```bash
wget https://files.pythonhosted.org/packages/45/35/65d9f8cc537129894b4b32647d80212d1fa342877581c5b8a69872cea8be/python-snappy-0.5.4.tar.gz
tar xzvf python-snappy-0.5.4.tar.gz
cd python-snappy-0.5.4
```

--------------------------------

### Send Message with Key for Partitioning

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Demonstrates sending a message with a key. Messages with the same key are hashed to the same partition. Keys and values should be raw bytes unless a serializer is configured.

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

--------------------------------

### List Partition Offsets

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to list offsets for partitions based on specific criteria.

```default
kafka-python admin partitions list-offsets [-h] [-t TOPIC] [-s SPEC]
                                           [-p PARTITIONS]
```

--------------------------------

### KafkaAdminClient.create_partitions

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaAdminClient.rst

Create additional partitions for an existing topic.

```APIDOC
## KafkaAdminClient.create_partitions(topic_partitions, timeout_ms=None, validate_only=False, raise_errors=True)

### Description
Create additional partitions for an existing topic.

### Parameters
- **topic_partitions** (dict) - Required - A dict of topic name strings to total partition count (int), or a dict of {topic_name: {count: int, assignments: [[broker_ids]]}} if manual assignment is desired.
- **timeout_ms** (numeric) - Optional - Milliseconds to wait for new partitions to be created before the broker returns.
- **validate_only** (bool) - Optional - If True, don’t actually create new partitions. Default: False.
- **raise_errors** (bool) - Optional - Whether to raise errors as exceptions. Default: True.

### Returns
- **Response** (CreatePartitionsResponse) - Appropriate version of CreatePartitionsResponse class.
```

--------------------------------

### Run integration tests

Source: https://github.com/dpkp/kafka-python/blob/master/docs/tests.rst

Manage Kafka broker binaries and execute integration tests requiring a Java environment.

```bash
# Download new broker files
KAFKA_VERSION=4.3.0 make servers/4.3.0/kafka-bin
# Run tests for previously-installed broker version
KAFKA_VERSION=4.3.0 pytest -v test/integration/
# Or install + run all tests
KAFKA_VERSION=4.3.0 make test
```

--------------------------------

### Produce messages with KafkaProducer

Source: https://github.com/dpkp/kafka-python/blob/master/docs/high_level.rst

Demonstrates basic message production, including synchronous blocking, keyed messages, custom serialization, and asynchronous callbacks.

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

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

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

# Block for 'synchronous' sends
try:
    record_metadata = future.get(timeout=10)
except KafkaError:
    # Decide what to do if produce request failed...
    log.exception()
    pass

# Successful result returns assigned partition and offset
print (record_metadata.topic)
print (record_metadata.partition)
print (record_metadata.offset)

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

# encode str with utf-8 encoding
producer = KafkaProducer(value_serializer=DefaultSerializer())
producer.send('utf-8-topic', 'value_str')

# encode str with utf-16 encoding
producer = KafkaProducer(value_serializer=DefaultSerializer('utf-16'))
producer.send('utf-16-topic', '懂不懂')

# produce json messages
producer = KafkaProducer(value_serializer=JsonSerializer())
producer.send('json-topic', {'key': 'value'})

# produce asynchronously
for _ in range(100):
    producer.send('my-topic', b'msg')

def on_send_success(record_metadata):
    print(record_metadata.topic)
    print(record_metadata.partition)
    print(record_metadata.offset)

def on_send_error(excp):
    log.error('I am an errback', exc_info=excp)
    # handle exception

# produce asynchronously with callbacks
producer.send('my-topic', b'raw_bytes').add_callback(on_send_success).add_errback(on_send_error)

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

# configure multiple retries
producer = KafkaProducer(retries=5)
```

--------------------------------

### CLI Usage Overview

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

General usage pattern for the admin cluster command.

```default
usage: kafka-python admin cluster [-h] COMMAND ...
```

--------------------------------

### Describe ACLs Command

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Displays the usage help for describing existing Kafka ACLs.

```default
kafka-python admin acls describe [-h] [--principal PRINCIPAL] [--host HOST]
                                 [--operation {unknown,any,all,read,write,create,delete,alter,describe,cluster_action,describe_configs,alter_configs,idempotent_write,create_tokens,describe_tokens}]
                                 [--permission-type {unknown,any,deny,allow}]
                                 [--resource-type {unknown,any,topic,group,cluster,transactional_id,delegation_token,user}]
                                 [--resource-name RESOURCE_NAME]
                                 [--pattern-type {unknown,any,match,literal,prefixed}]
```

--------------------------------

### Compress Messages with Gzip

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Demonstrates how to enable message compression using the 'gzip' type. This reduces message size on the wire. For other compression types, optional dependencies may be required.

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

--------------------------------

### Kafka-Python Admin Topics Usage

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Displays the base usage pattern for the admin topics command.

```default
usage: kafka-python admin topics [-h] COMMAND ...
```

--------------------------------

### Admin Partitions Usage

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

General usage syntax for the admin partitions command.

```default
usage: kafka-python admin partitions [-h] COMMAND ...
```

--------------------------------

### KafkaProducer Configuration Parameters

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaProducer.rst

A list of configuration parameters used to initialize the KafkaProducer client.

```APIDOC
## KafkaProducer Configuration

### Parameters
- **api_version** (tuple) - Optional - Specify which Kafka API version to use. If None, the client infers the broker version.
- **bootstrap_timeout_ms** (int) - Optional - Number of milliseconds to wait for the first successful cluster bootstrap. Default: 30000.
- **metric_reporters** (list) - Optional - A list of classes to use as metrics reporters. Default: [].
- **metrics_enabled** (bool) - Optional - Whether to track metrics on this instance. Default: True.
- **metrics_num_samples** (int) - Optional - The number of samples maintained to compute metrics. Default: 2.
- **metrics_sample_window_ms** (int) - Optional - The maximum age in milliseconds of samples used to compute metrics. Default: 30000.
- **net** (str or kafka.net.backend.NetBackend) - Optional - The async backend that runs this client’s network I/O event loop.
- **sasl_mechanism** (str) - Optional - Authentication mechanism (PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512).
- **sasl_plain_username** (str) - Optional - Username for sasl PLAIN and SCRAM authentication.
- **sasl_plain_password** (str) - Optional - Password for sasl PLAIN and SCRAM authentication.
- **sasl_kerberos_name** (str or gssapi.Name) - Optional - Constructed gssapi.Name for use with sasl mechanism handshake.
- **sasl_kerberos_service_name** (str) - Optional - Service name to include in GSSAPI sasl mechanism handshake. Default: 'kafka'.
- **sasl_kerberos_domain_name** (str) - Optional - Kerberos domain name to use in GSSAPI sasl mechanism handshake.
- **sasl_oauth_token_provider** (kafka.net.sasl.oauth.AbstractTokenProvider) - Optional - OAuthBearer token provider instance.
- **proxy_url** (str) - Optional - URL to proxy socket connections through (SOCKS5 only).
- **kafka_client** (callable) - Optional - Custom class/callable for creating KafkaNetClient instances.
```

--------------------------------

### Kafka-Python Admin Configs Usage

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Displays the base usage pattern for the admin configs command.

```default
usage: kafka-python admin configs [-h] COMMAND ...
```

--------------------------------

### Use KafkaProducer as a context manager

Source: https://github.com/dpkp/kafka-python/blob/master/docs/upgrade_to_3_0.rst

KafkaProducer, KafkaConsumer, and KafkaAdminClient now support the with statement for automatic resource management.

```python
with KafkaProducer(bootstrap_servers=...) as producer:
    producer.send('my-topic', b'hello')
```

--------------------------------

### Alter Log Directories

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to move replicas between log directories on hosting brokers.

```default
kafka-python admin cluster alter-log-dirs [-h] -a ASSIGNMENTS
```

--------------------------------

### Create ACLs Command

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Displays the usage help for creating new Kafka ACLs.

```default
kafka-python admin acls create [-h] --principal PRINCIPAL [--host HOST]
                               --operation
                               {unknown,any,all,read,write,create,delete,alter,describe,cluster_action,describe_configs,alter_configs,idempotent_write,create_tokens,describe_tokens}
                               [--permission-type {unknown,any,deny,allow}]
                               --resource-type
                               {unknown,any,topic,group,cluster,transactional_id,delegation_token,user}
                               --resource-name RESOURCE_NAME
                               [--pattern-type {unknown,any,match,literal,prefixed}]
```

--------------------------------

### seek_to_beginning(*partitions)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaConsumer.rst

Seek to the oldest available offset for partitions.

```APIDOC
## seek_to_beginning(*partitions)

### Description
Seek to the oldest available offset for partitions.

### Parameters
- **partitions** (TopicPartition) - Optional - Specific TopicPartitions to seek, defaults to all assigned.
```

--------------------------------

### kafka.KafkaAdminClient

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaAdminClient.rst

Initializes a new KafkaAdminClient instance for cluster administration.

```APIDOC
## class kafka.KafkaAdminClient(**configs)

### Description
A class for administering the Kafka cluster.

### Parameters
- **bootstrap_servers** (str or list) - Optional - 'host[:port]' string (or list of 'host[:port]' strings) that the consumer should contact to bootstrap initial cluster metadata.
- **client_id** (str) - Optional - A name for this client.
- **reconnect_backoff_ms** (int) - Optional - The amount of time in milliseconds to wait before attempting to reconnect to a given host.
- **reconnect_backoff_max_ms** (int) - Optional - The maximum amount of time in milliseconds to backoff/wait when reconnecting to a broker.
- **socket_connection_setup_timeout_ms** (int) - Optional - The maximum amount of time the client will wait for the socket connection to be established.
- **socket_connection_setup_timeout_max_ms** (int) - Optional - The maximum amount of time the client will wait for the socket connection to be established with exponential backoff.
- **request_timeout_ms** (int) - Optional - Client request timeout in milliseconds.
- **default_api_timeout_ms** (int) - Optional - Default timeout in milliseconds for blocking client APIs that do not take an explicit timeout.
- **connections_max_idle_ms** (int) - Optional - Close idle connections after the number of milliseconds specified.
- **retry_backoff_ms** (int) - Optional - Milliseconds to backoff when retrying on errors.
- **max_in_flight_requests_per_connection** (int) - Optional - Requests are pipelined to kafka brokers up to this number of maximum requests per broker connection.
- **receive_message_max_bytes** (int) - Optional - Maximum allowed network frame size.
- **receive_buffer_bytes** (int) - Optional - The size of the TCP receive buffer (SO_RCVBUF).
- **send_buffer_bytes** (int) - Optional - The size of the TCP send buffer (SO_SNDBUF).
- **socket_options** (list) - Optional - List of tuple-arguments to socket.setsockopt.
- **metadata_max_age_ms** (int) - Optional - The period of time in milliseconds after which we force a refresh of metadata.
- **security_protocol** (str) - Optional - Protocol used to communicate with brokers (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL).
- **ssl_context** (ssl.SSLContext) - Optional - Pre-configured SSLContext for wrapping socket connections.
- **ssl_check_hostname** (bool) - Optional - Flag to configure whether SSL handshake should verify that the certificate matches the broker’s hostname.
- **ssl_cafile** (str) - Optional - Optional filename of CA file to use in certificate.
```

--------------------------------

### Define new topic configuration structure

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaAdminClient.rst

The structure for defining new topics when using the create_topics method.

```default
{topic_name: {num_partitions: int (default -1),
              replication_factor: int (default -1),
              assignments: {partition_id: [broker_ids]},
              configs: {key: value}}}
```

--------------------------------

### Serialize String Keys

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Illustrates how to serialize message keys as strings using `DefaultSerializer`. The `key_serializer` parameter is set during producer initialization.

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

--------------------------------

### Manage Kafka Topics and Consumer Groups

Source: https://github.com/dpkp/kafka-python/blob/master/docs/high_level.rst

Demonstrates administrative operations including topic creation, deletion, and consumer group inspection. Requires a running Kafka broker connection.

```python
from kafka import KafkaAdminClient

admin = KafkaAdminClient(bootstrap_servers=['broker1:1234'])

# create topics with defaults (requires kafka 2.4+)
admin.create_topics(['testtopic1'], timeout_ms=None, validate_only=False)
# create a new topic with details
new_topics = {
    'num_partitions': 1,
    'replication_factor': 1,
    'assignments': {0: [1]},                      # assign partition 0 to broker id 1
    'configs': {'max_message_bytes': '1000000'},  # set non-default configs
}
admin.create_topics(new_topics, timeout_ms=None, validate_only=False)

# delete a topic
admin.delete_topics(['testtopic1'])

# list consumer groups
print(admin.list_groups())

# get consumer group details
print(admin.describe_groups(['cft-plt-qa.connect']))

# get consumer group offset
print(admin.list_group_offsets(['cft-plt-qa.connect']))
```

--------------------------------

### Include Record Headers

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Demonstrates how to include record headers with messages. Headers are provided as a list of tuples, where each tuple contains a string key and bytes value.

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

--------------------------------

### Use Kafka Transactions

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Shows how to use Kafka transactions for atomic message production. This includes initializing transactions, beginning, committing, and aborting transactions.

```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
```

--------------------------------

### Subscribe to Topics via Regex

Source: https://github.com/dpkp/kafka-python/blob/master/docs/high_level.rst

Using regex patterns to subscribe to multiple topics dynamically.

```python
# Subscribe to a regex topic pattern
consumer = KafkaConsumer()
consumer.subscribe(pattern='^awesome.*')
```

--------------------------------

### KafkaConsumer Configuration Patterns

Source: https://github.com/dpkp/kafka-python/blob/master/docs/high_level.rst

Various initialization patterns for offset resetting, deserialization, and timeouts.

```python
# consume earliest available messages, don't commit offsets
KafkaConsumer(auto_offset_reset='earliest', enable_auto_commit=False)

# consume json messages
KafkaConsumer(value_deserializer=JsonSerializer())

# consume utf-8
KafkaConsumer(value_deserializer=DefaultSerializer())

# consume utf-16
KafkaConsumer(value_deserializer=DefaultSerializer('utf-16'))

# StopIteration if no message after 1sec
KafkaConsumer(consumer_timeout_ms=1000)
```

--------------------------------

### Configure logging for kafka-python

Source: https://github.com/dpkp/kafka-python/blob/master/docs/index.rst

Enable internal operational event logging using the standard Python logging module.

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

--------------------------------

### KafkaAdminClient.describe_configs

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaAdminClient.rst

Fetch configuration parameters for one or more Kafka resources.

```APIDOC
## KafkaAdminClient.describe_configs(config_resources, include_synonyms=False, config_filter='modified')

### Description
Fetch configuration parameters for one or more Kafka resources.

### Parameters
- **config_resources** (list of ConfigResource) - Required - A list of ConfigResource objects.
- **include_synonyms** (bool) - Optional - If True, return synonyms in response. Default: False.
- **config_filter** (ConfigFilterType or str) - Optional - Modified returns only keys that have non-default values; Dynamic returns all keys that can be modified with alter_configs; All returns all available keys. Default: Modified.

### Returns
- **dict** - {resource_type (str): {resource_name (str): {config_key: {config data}}}}
```

--------------------------------

### Pass extra configuration to Kafka clients

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/index.rst

Use the -C flag to pass arbitrary keyword arguments to the underlying Kafka client constructor.

```bash
kafka-python consumer -b localhost:9092 -t foo \
    -C auto_offset_reset=earliest \
    -C consumer_timeout_ms=1000
```

--------------------------------

### KafkaAdminClient.describe_topic_partitions

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaAdminClient.rst

Describes topics with fine-grained partition-level control, supporting pagination and ELR information.

```APIDOC
## KafkaAdminClient.describe_topic_partitions(topics, response_partition_limit=2000, cursor=None)

### Description
Describes topics with fine-grained partition-level control (KIP-966). This method supports pagination via a cursor and provides partition-level ELR information.

### Parameters
- **topics** (list[str]) - Required - A list of topic names.
- **response_partition_limit** (int) - Optional - Maximum number of partitions to include in the response. Default: 2000.
- **cursor** (dict) - Optional - Dict with 'topic_name' and 'partition_index' keys to start pagination from. Default: None.

### Returns
- **dict** - Returns a dictionary containing 'topics' (list of topic details) and 'next_cursor' (pagination state).
```

--------------------------------

### Join a consumer group

Source: https://github.com/dpkp/kafka-python/blob/master/docs/index.rst

Configure a group_id to enable dynamic partition assignment and automatic offset commits.

```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)
```

--------------------------------

### Basic Kafka Consumer Usage

Source: https://github.com/dpkp/kafka-python/blob/master/README.rst

Initialize a KafkaConsumer to read messages from a specified topic. Iterates over messages and prints them.

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

--------------------------------

### List Partition Reassignments

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Command to view current ongoing partition reassignments.

```default
kafka-python admin partitions list-reassignments [-h] [-p TOPIC_PARTITIONS]
                                                 [--timeout-ms TIMEOUT_MS]
```

--------------------------------

### Send messages with keys

Source: https://github.com/dpkp/kafka-python/blob/master/docs/index.rst

Use keys to ensure messages with the same key are routed to the same partition.

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

--------------------------------

### List Config Resources

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Lists configuration resources known to the cluster. Requires broker version 4.1 or higher for non-client_metrics types.

```default
kafka-python admin configs list [-h] [-r RESOURCE_TYPES]
```

--------------------------------

### kafka.KafkaAdminClient(**configs)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaAdminClient.rst

Constructor for the KafkaAdminClient, used to initialize a connection to a Kafka cluster with specific configuration parameters.

```APIDOC
## class kafka.KafkaAdminClient(**configs)

### Description
Initializes a new KafkaAdminClient instance. This client provides a synchronous, blocking API for interacting with a Kafka cluster.

### Parameters
- **ssl_certfile** (str) - Optional - Filename of file in PEM format containing the client certificate.
- **ssl_keyfile** (str) - Optional - Filename containing the client private key.
- **ssl_password** (str) - Optional - Password for the certificate chain.
- **ssl_crlfile** (str) - Optional - Filename containing the CRL for certificate expiration checks.
- **api_version** (tuple) - Optional - Kafka API version to use (e.g., (2, 7)).
- **bootstrap_timeout_ms** (int) - Optional - Milliseconds to wait for cluster bootstrap. Default: 30000.
- **net** (str or NetBackend) - Optional - The async backend for network I/O.
- **metrics** (Metrics) - Optional - Metrics instance for capturing network IO stats.
- **metric_group_prefix** (str) - Optional - Prefix for metric names.
- **sasl_mechanism** (str) - Optional - Authentication mechanism (PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512).
- **sasl_plain_username** (str) - Optional - Username for PLAIN/SCRAM authentication.
- **sasl_plain_password** (str) - Optional - Password for PLAIN/SCRAM authentication.
- **sasl_kerberos_name** (str or gssapi.Name) - Optional - GSSAPI name for handshake.
- **sasl_kerberos_service_name** (str) - Optional - Service name for GSSAPI. Default: 'kafka'.
- **sasl_kerberos_domain_name** (str) - Optional - Kerberos domain name for GSSAPI.
- **sasl_oauth_token_provider** (AbstractTokenProvider) - Optional - OAuthBearer token provider.
- **proxy_url** (str) - Optional - SOCKS5 proxy URL.
- **kafka_client** (callable) - Optional - Custom callable for creating KafkaNetClient instances.
```

--------------------------------

### KafkaAdminClient.create_topics(new_topics, ...)

Source: https://github.com/dpkp/kafka-python/blob/master/docs/apidoc/KafkaAdminClient.rst

Creates new topics in the Kafka cluster.

```APIDOC
## KafkaAdminClient.create_topics(new_topics, timeout_ms=None, validate_only=False, raise_errors=True, wait_for_metadata=False)

### Description
Create new topics in the cluster.

### Parameters
- **new_topics** (list or dict) - A list of topic names, or a dict mapping each topic name to a dict of options.
- **timeout_ms** (numeric, optional) - Milliseconds to wait for new topics to be created.
- **validate_only** (bool, optional) - If True, don't actually create new topics. Default: False.
- **raise_errors** (bool, optional) - Whether to raise errors as exceptions. Default: True.
- **wait_for_metadata** (bool, optional) - If True, block until each new topic is visible in broker metadata. Default: False.
```

--------------------------------

### Kafka-python Admin Transactions Usage

Source: https://github.com/dpkp/kafka-python/blob/master/docs/cli/admin.rst

Displays the base usage pattern for the admin transactions command.

```default
usage: kafka-python admin transactions [-h] COMMAND ...
```