### Setup Python Test Environment

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Create and activate a virtual environment, then install test dependencies and the project itself.

```bash
python3 -m venv venv_test
source venv_test/bin/activate
python3 -m pip install -r requirements/requirements-tests-install.txt
python3 -m pip install .
```

```bash
deactivate
```

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

### Asynchronous Avro Serialization and Production Setup

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Example of setting up an asynchronous Avro producer. This involves instantiating an `AsyncSchemaRegistryClient`, an `AsyncAvroSerializer`, and then using the `AIOProducer` to produce messages. Ensure `user_schema` and `user_data` are defined.

```python
# From examples/README.md
from confluent_kafka.aio import AIOProducer
from confluent_kafka.schema_registry import AsyncSchemaRegistryClient
from confluent_kafka.schema_registry._async.avro import AsyncAvroSerializer

async def setup_async_avro_producer():
    # Use the appropriate configuration from the section above
    schema_registry_conf = {"url": "http://localhost:8081"}
    schema_registry_client = AsyncSchemaRegistryClient(schema_registry_conf)

    avro_serializer = await AsyncAvroSerializer(
        schema_registry_client=schema_registry_client,
        schema_str=user_schema
    )

    producer = AIOProducer({"bootstrap.servers": "localhost:9092"})

    # Serialize and produce
    serialized_data = await avro_serializer(user_data, SerializationContext("users", MessageField.VALUE))
    delivery_future = await producer.produce("users", value=serialized_data)
    message = await delivery_future
```

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

### Install via pip

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/INSTALL.md

Recommended method for installing pre-built wheels with all dependencies included.

```bash
python3 -m pip install confluent-kafka
```

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

### Setup venv with Release Version

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Installs the latest release version of confluent-kafka and example dependencies into a virtual environment. Ensure you have Python 3 installed.

```bash
python3 -m venv venv_examples
source venv_examples/bin/activate
python3 -m pip install confluent_kafka
python3 -m pip  install -r requirements/requirements-examples.txt
```

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

### Local Setup with UV (Python 3.11)

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Alternative setup using `uv` for Python 3.11, including virtual environment creation, dependency synchronization, and testing.

```bash
# Modify pyproject.toml to require python version >=3.11
# This fixes the cel-python dependency conflict
uv venv --python 3.11
source .venv/bin/activate

uv sync --extra dev --extra tests
uv pip install trivup setuptools
pytest tests/

# When making changes, change project.version in pyproject.toml before re-running:
uv sync --extra dev --extra tests
```

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

### Install Documentation Dependencies

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Install the necessary Python packages required for building the project's documentation.

```bash
pip3 install .[docs]
```

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

### Basic Producer/Consumer Command Line Usage

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Illustrates the basic command-line pattern for running producer and consumer examples. Replace `<example_name>.py` with the specific example script and provide necessary arguments like bootstrap servers.

```bash
# Most examples follow this pattern:
python3 <example_name>.py <bootstrap_servers> [additional_args]

# For example:
python3 producer.py localhost:9092
python3 consumer.py localhost:9092
python3 asyncio_example.py localhost:9092 my-topic
```

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

### Verify Setup

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Run a simple Python command to verify that the confluent_kafka library has been imported successfully, indicating a successful setup.

```python
python3 -c "import confluent_kafka; print('Setup successful!')"
```

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

### Schema Registry Example Command Line Usage

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Shows the command-line pattern for running examples that require a Schema Registry, such as Avro serialization. Ensure a Schema Registry is running and accessible.

```bash
python3 avro_producer.py localhost:9092 http://localhost:8081
```

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

### Install from source on Debian or Ubuntu

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/INSTALL.md

Requires root privileges to install build tools and configure the Confluent repository.

```bash
#
# Perform these steps as the root user (e.g., in a 'sudo bash' shell)
#

# Install build tools and Kerberos support.

apt install -y wget software-properties-common lsb-release gcc make python3 python3-pip python3-dev libsasl2-modules-gssapi-mit krb5-user


# Install the latest version of librdkafka:

wget -qO - https://packages.confluent.io/deb/7.0/archive.key | apt-key add -

add-apt-repository "deb https://packages.confluent.io/clients/deb $(lsb_release -cs) main"

apt update

apt install -y librdkafka-dev


#
# Now build and install confluent-kafka-python as your standard user
# (e.g., exit the root shell first).
#

python3 -m pip install --no-binary confluent-kafka confluent-kafka


# Verify that confluent_kafka is installed:

python3 -c 'import confluent_kafka; print(confluent_kafka.version())'
```

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

### Install Tox

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Install the tox testing tool using pip.

```bash
pip install tox
```

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

### Install from source on Mac OS X

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/INSTALL.md

Uses Homebrew to install librdkafka before building the Python package.

```bash

# Install librdkafka from homebrew

brew install librdkafka


# Build and install confluent-kafka-python

python3 -m pip install --no-binary confluent-kafka confluent-kafka


# Verify that confluent_kafka is installed:

python3 -c 'import confluent_kafka; print(confluent_kafka.version())'
```

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

### Setup venv with Development Version

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Installs the current source tree version of confluent_kafka into a virtual environment. Requires a C compiler and librdkafka. Ensure you have Python 3 installed.

```bash
python3 -m venv venv_examples
source venv_examples/bin/activate
python3 -m pip install .[examples]
```

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

### Install and Run Code Formatting Tools

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/CONTRIBUTOR.md

Install black, isort, and flake8 for code style checks and formatting. Use 'make style-check' to verify and 'make style-fix' to automatically correct.

```bash
pip install black isort flake8

make style-check

make style-fix
```

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

### Install confluent-kafka-python with Development Extras

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Install the library in editable mode with development, testing, and documentation dependencies. This allows for direct code changes and testing.

```bash
pip3 install -e .[dev,tests,docs]
```

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

### Install confluent-kafka package

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Provides various pip installation commands for the base library and optional support for Schema Registry and Data Contract rules.

```bash
# Basic installation
pip install confluent-kafka

# With Schema Registry support
pip install "confluent-kafka[avro,schemaregistry]"     # Avro
pip install "confluent-kafka[json,schemaregistry]"     # JSON Schema
pip install "confluent-kafka[protobuf,schemaregistry]" # Protobuf

# With Data Contract rules (includes CSFLE support)
pip install "confluent-kafka[avro,schemaregistry,rules]"
```

```bash
pip install "confluent-kafka[avro,schemaregistry]"
```

```bash
pip install "confluent-kafka[json,schemaregistry]"
```

```bash
pip install "confluent-kafka[protobuf,schemaregistry]"
```

```bash
pip install "confluent-kafka[avro,schemaregistry,rules]"
```

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

### Install from source on RedHat, CentOS, Fedora

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/INSTALL.md

Requires root privileges for system dependencies and building from source for specific features like Kerberos.

```bash
#
# Perform these steps as the root user (e.g., in a 'sudo bash' shell)
#

# Install build tools and Kerberos support.

yum install -y python3 python3-pip python3-devel gcc make cyrus-sasl-gssapi krb5-workstation

# Install the latest version of librdkafka:

rpm --import https://packages.confluent.io/rpm/7.0/archive.key

echo '
[Confluent-Clients]
name=Confluent Clients repository
baseurl=https://packages.confluent.io/clients/rpm/centos/$releasever/$basearch
gpgcheck=1
gpgkey=https://packages.confluent.io/clients/rpm/archive.key
enabled=1' > /etc/yum.repos.d/confluent.repo

yum install -y librdkafka-devel


#
# Now build and install confluent-kafka-python as your standard user
# (e.g., exit the root shell first).
#

python3 -m pip install --no-binary confluent-kafka confluent-kafka


# Verify that confluent_kafka is installed:

python3 -c 'import confluent_kafka; print(confluent_kafka.version())'
```

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

### Create and Activate Virtual Environment

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Create a Python virtual environment for isolating project dependencies. Activate it before proceeding with installations.

```bash
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
```

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

### Verify Python Client Setup with Ducktape

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/kafkatest/README.md

Run a specific test to verify the Python client setup using ducktape. Check the test log for output indicating successful execution.

```bash
cd $KAFKA_DIR
ducktape --globals tests/confluent-kafka-python/globals.json tests/kafkatest/tests/client/pluggable_test.py
```

```bash
grep confluent-kafka-python results/latest/PluggableConsumerTest/test_start_stop/1/test_log.debug
```

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

### Start Kafka Cluster for Old Integration Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Bring up a Kafka cluster and Schema Registry instances using the provided Docker script. This is required for the older integration tests.

```bash
./tests/docker/bin/cluster_up.sh
```

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

### AsyncIO Producer with Schema Registry and Avro Serialization

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Example of setting up an AIOProducer with AsyncSchemaRegistryClient and AsyncAvroSerializer for producing Avro-formatted messages. Ensure the Schema Registry URL and topic schema are correctly configured.

```python
from confluent_kafka.aio import AIOProducer
from confluent_kafka.schema_registry import AsyncSchemaRegistryClient
from confluent_kafka.schema_registry._async.avro import AsyncAvroSerializer

async def setup_async_avro_producer():
    schema_registry_client = AsyncSchemaRegistryClient({"url": "http://localhost:8081"})

    avro_serializer = await AsyncAvroSerializer(
        schema_registry_client=schema_registry_client,
        schema_str=user_schema
    )

    producer = AIOProducer({"bootstrap.servers": "localhost:9092"})

    # Serialize and produce
    serialized_data = await avro_serializer(user_data, SerializationContext("users", MessageField.VALUE))
    delivery_future = await producer.produce("users", value=serialized_data)
    message = await delivery_future
```

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

### Specify librdkafka Installation Path

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

When librdkafka is installed in a non-standard location, provide the include and library paths using C_INCLUDE_PATH and LIBRARY_PATH environment variables during the build process.

```bash
C_INCLUDE_PATH=/path/to/include LIBRARY_PATH=/path/to/lib python -m build
```

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

### Synchronous Avro Serialization and Production

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Example of configuring a synchronous Avro serializer and producing messages. The serializer handles schema registration and caching. Ensure `user_schema` and `some_user_object` are defined.

```python
# From examples/avro_producer.py
schema_registry_conf = {'url': 'http://localhost:8081'}
# For Confluent Cloud you can pass `--sr-api-key` and `--sr-api-secret` to the script.
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

avro_serializer = AvroSerializer(schema_registry_client,
                                 user_schema,
                                 lambda user, ctx: user.to_dict())

producer_conf = {
    'bootstrap.servers': 'localhost:9092',
}

producer = Producer(producer_conf)

# Simplified example of serialize and produce
serialized_value = avro_serializer(some_user_object)
producer.produce('my-topic', key='user1', value=serialized_value)
producer.flush()
```

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

### Unit Test Structure Example

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/CONTRIBUTOR.md

A standard Python unit test structure using Arrange-Act-Assert pattern. Ensure all new functionality includes corresponding unit tests.

```python
def test_feature_should_behave_correctly_when_condition():
    # Arrange
    setup_data = create_test_data()
    
    # Act
    result = function_under_test(setup_data)
    
    # Assert
    assert result.expected_property == expected_value
```

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

### Build the Bundle Independently

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Build the confluent-kafka-python package bundle without an editable install. This is useful for creating distributable artifacts.

```bash
python3 -m build
```

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

### Classic Protocol Rebalance Callback

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/docs/kip-848-migration-guide.md

Example rebalance callback functions for the classic Kafka protocol's range assignor. The `partitions` list contains the full partition list.

```python
# Rebalance Callback for Range Assignor (Classic Protocol)
def on_assign(consumer, partitions):
    # Full partition list is provided under the classic protocol
    print(f"[Classic] Assigned partitions: {partitions}")
    consumer.assign(partitions) # Optional: client handles if not used

def on_revoke(consumer, partitions):
    print(f"[Classic] Revoked partitions: {partitions}")
    consumer.unassign() # Optional: client handles if not used
```

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

### Run Specific Integration Tests with Pytest

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Execute a specific integration test file using pytest. This utilizes the 'kafka_cluster' fixture for automatic cluster setup.

```bash
pytest -v -s ./tests/integration/consumer/test_consumer_error.py
```

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

### Next-Gen Protocol Incremental Rebalance Callback

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/docs/kip-848-migration-guide.md

Example rebalance callback functions for the next-generation Kafka protocol's incremental assignor. The `partitions` list contains only incrementally assigned or revoked partitions.

```python
# Rebalance callback for incremental assignor
def on_assign(consumer, partitions):
    # Only incremental partitions are passed here (not full list)
    print(f"[KIP-848] Incrementally assigning: {partitions}")
    consumer.incremental_assign(partitions) # Optional: client handles if not used

def on_revoke(consumer, partitions):
    print(f"[KIP-848] Incrementally revoking: {partitions}")
    consumer.incremental_unassign(partitions) # Optional: client handles if not used
```

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

### Build HTML Documentation with Setup.py

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Generate the project's HTML documentation using `setup.py`. The output will be located in the `build/sphinx/html` directory.

```bash
python3 setup.py build_sphinx
```

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

### Initialize Kafka Consumer with Configuration

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/docs/index.md

Demonstrates how to create a Kafka consumer instance by providing a dictionary of configuration properties, including bootstrap servers, group ID, session timeout, and offset reset behavior.

```python
conf = {'bootstrap.servers': 'mybroker.com',
        'group.id': 'mygroup',
        'session.timeout.ms': 6000,
        'on_commit': my_commit_callback,
        'auto.offset.reset': 'earliest'}
consumer = confluent_kafka.Consumer(conf)
```

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

### Build HTML Documentation with Make

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Generate the project's HTML documentation using the `make docs` command. The output will be located in the `docs/_build/` directory.

```bash
make docs
```

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

### Run Full Kafka System-Test Suite

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/kafkatest/README.md

Execute the complete Kafka system-test client suite using ducktape.

```bash
cd $KAFKA_DIR
ducktape --globals tests/confluent-kafka-python/globals.json tests/kafkatest/tests/client
```

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

### Bootstrap the soak testing environment

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/soak/README.md

Initializes the environment with specific Python and librdkafka versions.

```bash
./bootstrap.sh <python-branch/tag> <librdkafka-branch/tag>
```

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

### Run Ducktape Tests with Performance Metrics

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/ducktape/README.md

Execute all Ducktape tests with integrated performance metrics. Ensure Kafka and Schema Registry are running as per prerequisites.

```bash
./tests/ducktape/run_ducktape_test.py
```

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

### Prepare Deploy Directory for Python Client

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/kafkatest/README.md

Copy Python wheels and a deploy script to a location accessible by kafkatest worker instances. Synchronize the shared directory with workers using vagrant rsync.

```bash
confluent_kafka/kafkatests/deploy.sh --prepare $KAFKA_DIR wheels
```

```bash
cd $KAFKA_DIR
vagrant rsync
```

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

### FastAPI Integration with AIOProducer

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Demonstrates how to integrate AIOProducer with FastAPI for asynchronous event handling. Ensure the producer is initialized on application startup.

```python
from fastapi import FastAPI
from confluent_kafka.aio import AIOProducer

app = FastAPI()
producer = None

@app.on_event("startup")
async def startup():
    global producer
    producer = AIOProducer({"bootstrap.servers": "localhost:9092"})

@app.post("/events")
async def create_event(data: dict):
    delivery_future = await producer.produce("events", value=str(data))
    message = await delivery_future
    return {"offset": message.offset()}
```

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

### Build Docker Image with Dockerfile.alpine

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/docker/README.md

Use this command to build a Docker image using the Dockerfile.alpine. Ensure you are in the source top directory.

```bash
docker build -f examples/docker/Dockerfile.alpine .

```

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

### Configure Schema Registry Client

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Configuration for connecting to Schema Registry, shown for both local Confluent Platform and Confluent Cloud.

```python
# For Confluent Platform (local)
schema_registry_conf = {'url': 'http://localhost:8081'}

# For Confluent Cloud
# schema_registry_conf = {
#     'url': 'https://<ccloud-schema-registry-url>',
#     'basic.auth.user.info': '<sr-api-key>:<sr-api-secret>'
# }
```

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

### Initialize MetricsBounds in Python

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/ducktape/README.md

Load performance bounds using the MetricsBounds class. It can automatically load from environment variables or a specified config file.

```python
from producer_benchmark_metrics import MetricsBounds

# Loads from BENCHMARK_BOUNDS_CONFIG env var, or producer_benchmark_bounds.json if not set
bounds = MetricsBounds()

# Or load from a specific config file
bounds = MetricsBounds.from_config_file("my_bounds.json")
```

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

### Initialize Release Candidate Branch

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tools/RELEASE.md

Commands to ensure the local repository is synchronized with the master branch and to create a new release candidate branch.

```bash
$ git checkout master
$ git pull --rebase --tags origin master

Create an RC branch

     $ git checkout -b v0.11.4rc
```

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

### Run Formatting Checks and Fixes

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Use make commands to verify or automatically correct code style.

```bash
# Check formatting
make style-check

# Fix formatting
make style-fix

# Check only changed files
make style-check-changed
make style-fix-changed
```

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

### Configure Classic Protocol

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/docs/index.md

Configuration settings for the classic Kafka consumer protocol.

```properties
# Optional; default is 'classic'
group.protocol=classic

partition.assignment.strategy=<range,roundrobin,sticky>
session.timeout.ms=45000
heartbeat.interval.ms=15000
```

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

### Run Subset of Kafka System-Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/kafkatest/README.md

Execute a specific subset of Kafka system-tests using ducktape, targeting a particular test class.

```bash
cd $KAFKA_DIR
ducktape --globals tests/confluent-kafka-python/globals.json tests/kafkatest/tests/client/consumer_test.py::AssignmentValidationTest
```

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

### Run Linting and Typing Checks with Tox

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Use tox environments to run automated linting, formatting, and type checking.

```bash
# Check formatting
tox -e black,isort

# Check linting
tox -e flake8

# Check typing
tox -e mypy

# Run all formatting and linting checks
tox -e black,isort,flake8,mypy
```

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

### Manage topics with AdminClient

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Uses AdminClient to programmatically create new Kafka topics.

```python
from confluent_kafka.admin import AdminClient, NewTopic

a = AdminClient({'bootstrap.servers': 'mybroker'})

new_topics = [NewTopic(topic, num_partitions=3, replication_factor=1) for topic in ["topic1", "topic2"]]
```

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

### Run Old Integration Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Execute the older integration tests using Python, specifying the test configuration file.

```bash
python3 ./tests/integration/integration_test.py ./tests/integration/testconf.json
```

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

### Asynchronously create Kafka topics

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Uses create_topics to initiate topic creation and iterates over the returned futures to handle results or exceptions.

```python
# Call create_topics to asynchronously create topics. A dict
# of <topic,future> is returned.
fs = a.create_topics(new_topics)

# Wait for each operation to finish.
for topic, f in fs.items():
    try:
        f.result()  # The result itself is None
        print("Topic {} created".format(topic))
    except Exception as e:
        print("Failed to create topic {}: {}".format(topic, e))
```

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

### Environment-Based Performance Bounds Configuration

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/ducktape/README.md

JSON configuration defining performance thresholds for different environments (e.g., local, CI). Use this to customize performance expectations.

```json
{
  "_comment": "Performance bounds for benchmark tests by environment",
  "local": {
    "_comment": "Default bounds for local development - more relaxed thresholds",
    "min_throughput_msg_per_sec": 1000.0,
    "max_p95_latency_ms": 2000.0,
    "max_error_rate": 0.02,
    "min_success_rate": 0.98,
    "max_p99_latency_ms": 3000.0,
    "max_memory_growth_mb": 1000.0,
    "max_buffer_full_rate": 0.05,
    "min_messages_per_poll": 10.0
  },
  "ci": {
    "_comment": "Stricter bounds for CI environment - production-like requirements",
    "min_throughput_msg_per_sec": 1500.0,
    "max_p95_latency_ms": 1500.0,
    "max_error_rate": 0.01,
    "min_success_rate": 0.99,
    "max_p99_latency_ms": 2500.0,
    "max_memory_growth_mb": 700.0,
    "max_buffer_full_rate": 0.03,
    "min_messages_per_poll": 15.0
  },
  "_default_environment": "local"
}
```

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

### Activate virtual environment

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/soak/README.md

Activates the Python virtual environment before running tests.

```bash
. venv/bin/activate
```

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

### Run All Tests with Tox

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Execute all configured tox environments, including integration tests, from the top-level directory.

```bash
./tests/run.sh tox
```

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

### Build the soak testing environment

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/soak/README.md

Updates the environment for subsequent runs using specified branches or tags.

```bash
./build.sh <python-branch/tag> <librdkafka-branch/tag>
```

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

### Configure Integration Tests to Use Existing Brokers

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Set the BROKERS environment variable to point to an existing Kafka cluster. SR_URL is optional and needed for Schema Registry tests.

```bash
export BROKERS=localhost:9092
# SR_URL is optional and only required for Schema-registry tests
export SR_URL=http://localhost:1234/
```

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

### Run Integration Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Execute integration tests which may require a local or CI Kafka cluster.

```bash
pytest -q tests/integration
```

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

### Clone Repository

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Clone the confluent-kafka-python repository from GitHub. This is the first step for setting up a local development environment.

```bash
git clone https://github.com/your-username/confluent-kafka-python.git
cd confluent-kafka-python
```

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

### Run Verifiable Consumer Standalone

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/kafkatest/README.md

Run the verifiable consumer directly using the Python module.

```python
python -m confluent_kafka.kafkatest.verifiable_consumer <options>
```

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

### Build Python Wheels

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/kafkatest/README.md

Build self-contained binary Linux wheels for the Python client and its dependencies. Ensure Python 2.7-x64 packages are built by using CIBW_SKIP.

```bash
CIBW_SKIP="cp3* *i686*" tools/cibuildwheel-build.sh wheels
```

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

### Run Unit Tests using run.sh Script

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Execute unit tests by invoking the provided run.sh script with the 'unit' argument.

```bash
./tests/run.sh unit
```

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

### Configure Tox for Integration Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Uncomment the line in tox.ini to enable running integration tests with tox.

```ini
#python3 tests/integration/integration_test.py
```

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

### Run Unit Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Execute the full suite of unit tests using pytest.

```bash
pytest -q
```

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

### Execute soak tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/soak/README.md

Runs the soak test suite using the provided configuration file and a unique test identifier.

```bash
TESTID=<testid> ./run.sh ccloud.config
```

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

### Legacy Producers and Consumers

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/docs/index.md

Notes on legacy AvroProducer and AvroConsumer classes.

```APIDOC
## Legacy Avro Producers and Consumers

### Description
These sections refer to legacy implementations of AvroProducer and AvroConsumer. It is recommended to use newer, more robust serialization methods if available.

- **AvroProducer (Legacy)**
- **AvroConsumer (Legacy)**
```

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

### aiohttp Integration with AIOProducer

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md

Shows how to integrate AIOProducer with aiohttp web applications. The producer is attached to the application instance for easy access within request handlers.

```python
from aiohttp import web
from confluent_kafka.aio import AIOProducer

async def init_app():
    app = web.Application()
    app['producer'] = AIOProducer({"bootstrap.servers": "localhost:9092"})
    return app

async def handle_event(request):
    producer = request.app['producer']
    delivery_future = await producer.produce("events", value="data")
    message = await delivery_future
    return web.json_response({"offset": message.offset()})
```

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

### Run Async Producer Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Execute specific tests related to the AIOProducer component.

```bash
pytest -q tests/test_AIOProducer.py
pytest -q -k AIOProducer
```

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

### Run Ducktape Tests by File

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/ducktape/README.md

Execute all Ducktape tests within a specific file. This is useful for focusing on a particular set of tests.

```bash
./tests/ducktape/run_ducktape_test.py producer
```

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

### Produce messages with synchronous Schema Registry

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Uses the synchronous SchemaRegistryClient and AvroSerializer to produce messages to a Kafka topic.

```python
from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import StringSerializer, SerializationContext, MessageField

# Configure Schema Registry Client
schema_registry_conf = {'url': 'http://localhost:8081'}  # Confluent Platform
# For Confluent Cloud, add: 'basic.auth.user.info': '<sr-api-key>:<sr-api-secret>'
# See: https://docs.clou.confluent.io/cloud/current/sr/index.html
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

# 2. Configure AvroSerializer
avro_serializer = AvroSerializer(schema_registry_client,
                                 user_schema_str,
                                 lambda user, ctx: user.to_dict())

# 3. Configure Producer
producer_conf = {
    'bootstrap.servers': 'localhost:9092',
}
producer = Producer(producer_conf)

# 4. Produce messages
serialized_value = avro_serializer(some_user_object)
producer.produce('my-topic', key='user1', value=serialized_value)
producer.flush()
```

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

### Run Verifiable Producer Standalone

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/kafkatest/README.md

Run the verifiable producer directly using the Python module.

```python
python -m confluent_kafka.kafkatest.verifiable_producer <options>
```

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

### Commit Changes

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/CONTRIBUTOR.md

Stage all changes and commit them with a clear, descriptive message. Follow guidelines for message format, length, and issue referencing.

```bash
git add .
git commit -m "Clear, descriptive commit message"
```

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

### Forward Kafka Client Logs to Python Logging Handler

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/docs/index.md

Shows how to integrate the Kafka client's logging with Python's built-in logging module by providing a logger instance. Logs are forwarded when client.poll() or producer.flush() are called.

```python
mylogger = logging.getLogger()
mylogger.addHandler(logging.StreamHandler())
producer = confluent_kafka.Producer({'bootstrap.servers': 'mybroker.com'}, logger=mylogger)
```

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

### Produce messages with asynchronous Schema Registry

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Uses AsyncSchemaRegistryClient and AsyncAvroSerializer with AIOProducer for non-blocking operations.

```python
from confluent_kafka.aio import AIOProducer
from confluent_kafka.schema_registry import AsyncSchemaRegistryClient
from confluent_kafka.schema_registry._async.avro import AsyncAvroSerializer

# Setup async Schema Registry client and serializer
# (See configuration options in the synchronous example above)
schema_registry_conf = {'url': 'http://localhost:8081'}
schema_client = AsyncSchemaRegistryClient(schema_registry_conf)
serializer = await AsyncAvroSerializer(schema_client, schema_str=avro_schema)

# Use with AsyncIO producer
producer = AIOProducer({"bootstrap.servers": "localhost:9092"})
serialized_value = await serializer(data, SerializationContext("topic", MessageField.VALUE))
delivery_future = await producer.produce("topic", value=serialized_value)
```

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

### Run Specific Ducktape Test with Metrics

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/ducktape/README.md

Execute a single, specific Ducktape test case and include performance metrics. This is ideal for debugging or analyzing a particular test.

```bash
./tests/ducktape/run_ducktape_test.py producer SimpleProducerTest.test_basic_produce
```

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

### OpenSSL Configuration for FIPS

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/fips/README.md

This configuration enables the FIPS provider and sets default properties for FIPS compliance. Ensure the `fipsmodule.cnf` path is correct for your system.

```ini
config_diagnostics = 1
openssl_conf = openssl_init

.include /usr/local/ssl/fipsmodule.cnf

[openssl_init]
providers = provider_sect
alg_section = algorithm_sect

[provider_sect]
fips = fips_sect

[algorithm_sect]
default_properties = fips=yes
.
```

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

### Set Environment Variables for Old Integration Tests

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tests/README.md

Source the .env.sh script to set environment variables required by the old integration tests, including those referenced in testconf.json.

```bash
source ./tests/docker/.env.sh
```

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

### Run FIPS Compliant Producer

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/fips/docker/README.md

Execute the FIPS compliant producer script. Ensure you set the OPENSSL_CONF and OPENSSL_MODULES environment variables to point to your FIPS-enabled OpenSSL configuration and modules.

```bash
OPENSSL_CONF="/path/to/fips/enabled/openssl/config/openssl.cnf" OPENSSL_MODULES="/path/to/fips/module/lib/folder/" ./examples/fips/fips_producer.py localhost:9092 test-topic
```

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

### Create a Feature Branch

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/CONTRIBUTOR.md

Use git to create a new branch for your feature or bug fix. Prefix feature branches with 'feature/' and bug fix branches with 'fix/'.

```bash
git checkout -b feature/your-feature-name
# or
git checkout -b fix/issue-number-description
```

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

### Run Unasync Script

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/DEVELOPER.md

Execute the `unasync.py` script to generate synchronous versions of asynchronous code, primarily for the Schema Registry components. This script is essential for maintaining consistency between async and sync APIs.

```bash
python3 tools/unasync.py
```

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

### Identify Changes for Changelog

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/tools/RELEASE.md

Command to list commits since the last release to identify necessary changelog entries.

```bash
$ git log v0.11.3..    # the last release
```

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

### Classic Protocol Configuration

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/docs/kip-848-migration-guide.md

Configuration properties for the classic Kafka consumer protocol. The default protocol is 'classic'.

```properties
group.protocol=classic

partition.assignment.strategy=<range,roundrobin,sticky>
session.timeout.ms=45000
heartbeat.interval.ms=15000
```

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

### Consume messages with basic Consumer

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Implements a basic consumer loop to poll and process messages from a Kafka topic.

```python
from confluent_kafka import Consumer

c = Consumer({
    'bootstrap.servers': 'mybroker',
    'group.id': 'mygroup',
    'auto.offset.reset': 'earliest'
})

c.subscribe(['mytopic'])

while True:
    msg = c.poll(1.0)

    if msg is None:
        continue
    if msg.error():
        print("Consumer error: {}".format(msg.error()))
        continue

    print('Received message: {}'.format(msg.value().decode('utf-8')))

c.close()
```

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

### Push Changes to Remote

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/CONTRIBUTOR.md

Push your local feature branch to the origin remote repository to prepare for creating a pull request.

```bash
git push origin feature/your-feature-name
```

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

### Import paths for async serializers

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Required import statements for accessing asynchronous serializers and deserializers.

```python
from confluent_kafka.schema_registry._async.avro import AsyncAvroSerializer, AsyncAvroDeserializer
from confluent_kafka.schema_registry._async.json_schema import AsyncJSONSerializer, AsyncJSONDeserializer
from confluent_kafka.schema_registry._async.protobuf import AsyncProtobufSerializer, AsyncProtobufDeserializer
```

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

### Produce messages using AsyncIO Producer

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Use AIOProducer in event-loop-based applications to avoid blocking. Note that per-message headers are not supported in the batched async path.

```python
import asyncio
from confluent_kafka.aio import AIOProducer

async def main():
    p = AIOProducer({"bootstrap.servers": "mybroker"})
    try:
        # produce() returns a Future; first await the coroutine to get the Future,
        # then await the Future to get the delivered Message.
        delivery_future = await p.produce("mytopic", value=b"hello")
        delivered_msg = await delivery_future
        # Optionally flush any remaining buffered messages before shutdown
        await p.flush()
    finally:
        await p.close()

asyncio.run(main())
```

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

### Produce messages using synchronous Producer

Source: https://github.com/confluentinc/confluent-kafka-python/blob/master/README.md

Use the standard Producer for scripts and high-throughput pipelines. Requires calling poll() or flush() to trigger delivery callbacks.

```python
from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'mybroker1,mybroker2'})

def delivery_report(err, msg):
    """ Called once for each message produced to indicate delivery result.
        Triggered by poll() or flush()."""
    if err is not None:
        print('Message delivery failed: {}'.format(err))
    else:
        print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))

for data in some_data_source:
    # Trigger any available delivery report callbacks from previous produce() calls
    p.poll(0)

    # Asynchronously produce a message. The delivery report callback will
    # be triggered from the call to poll() above, or flush() below, when the
    # message has been successfully delivered or failed permanently.
    p.produce('mytopic', data.encode('utf-8'), callback=delivery_report)

# Wait for any outstanding messages to be delivered and delivery report
# callbacks to be triggered.
p.flush()
```