<a id="python-client"></a>

# Python Client for Apache Kafka

## Overview

Confluent, a leading developer and maintainer of Apache Kafka®, offers
[confluent-kafka-python on GitHub](https://github.com/confluentinc/confluent-kafka-python). This Python client
provides a high-level producer, consumer, and AdminClient that are compatible
with Apache Kafka® brokers (version 0.8 or later), Confluent Cloud, and Confluent
Platform. Stay up-to-date with the latest release updates by checking out the
[changelog](https://github.com/confluentinc/confluent-kafka-python/blob/master/CHANGELOG.md)
available in the same repository.

<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

The confluent-kafka-python package is a binding on top of the C client,
[librdkafka](https://github.com/edenhill/librdkafka). For overview of the
librdkafka client library, see [Introduction to librdkafka client library](https://github.com/confluentinc/librdkafka/blob/master/INTRODUCTION.md).

For information about the configuration of the Confluent Kafka Python client, see
[Kafka client configuration](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#kafka-client-configuration).

<a id="installation-python-client"></a>

## Installation

The client is available on [PyPI](https://pypi.python.org/pypi/confluent-kafka) and can be installed using `pip`:

```bash
pip install confluent-kafka
```

You can install it globally, or within a [virtualenv](https://docs.python.org/3/library/venv.html). If you want to
install a FIPS-compliant client, see [FIPS compliance](#fips-compliance).

#### NOTE
The confluent-kafka-python package comes bundled with a pre-built version of
[librdkafka](https://github.com/edenhill/librdkafka) which does not
include GSSAPI/Kerberos support. For information about how to install a
version that supports GSSAPI, see the [installation instructions](https://github.com/confluentinc/confluent-kafka-python#install).

## Example code

For a step-by-step tutorial using the Python client including code samples for the producer and consumer see
[this guide](https://developer.confluent.io/get-started/python/).

For examples using basic producers, consumers, AsyncIO, and how to produce and consume Avro data with Schema Registry,
see [confluent-kafka-python GitHub repository](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md).

<a id="producer-python-client"></a>

## Kafka producer

### Initialization

The producer is configured using a dictionary in the examples below. If you are running Kafka locally,
you can initialize the producer as shown below.

```python
from confluent_kafka import Producer
import socket

conf = {'bootstrap.servers': 'host1:9092,host2:9092',
        'client.id': socket.gethostname()}

producer = Producer(conf)
```

If you are connecting to a Kafka cluster in Confluent Cloud, you must provide
credentials for access. The example below shows using a cluster API key and
secret.

```python
from confluent_kafka import Producer
import socket

conf = {'bootstrap.servers': 'pkc-abcd85.us-west-2.aws.confluent.cloud:9092',
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'PLAIN',
        'sasl.username': '<CLUSTER_API_KEY>',
        'sasl.password': '<CLUSTER_API_SECRET>',
        'client.id': socket.gethostname()}

producer = Producer(conf)
```

* For information on the available configuration properties, see the
  [API Documentation](/platform/current/clients/confluent-kafka-python/html/index.html).
* For a step-by-step tutorial using the Python client including code samples for
  the producer and consumer see [this guide](https://developer.confluent.io/get-started/python/).

### Asynchronous writes

To initiate sending a message to Kafka, call the `produce` method, passing in the
message value (which may be `None`) and optionally a key, partition, and callback.
The produce call completes immediately and does not return a value. A
`KafkaException` is thrown if the message could not be enqueued due to
librdkafka’s local produce queue being full.

```python
producer.produce(topic, key="key", value="value")
```

To receive notification of delivery success or failure, you can pass a `callback`
parameter. This can be any callable, for example, a lambda, function, bound method, or
callable object. Although the `produce()` method enqueues message immediately
for batching, compression and transmission to broker, no delivery notification events
are propagated until `poll()` is invoked.

```python
def acked(err, msg):
    if err is not None:
        print("Failed to deliver message: %s: %s" % (str(msg), str(err)))
    else:
        print("Message produced: %s" % (str(msg)))

producer.produce(topic, key="key", value="value", callback=acked)

# Wait up to one second for events. Callbacks will be invoked during
# this method call if the message is acknowledged.
producer.poll(1)
```

#### Perform rebalance with asynchronous callbacks

Define asynchronous callback functions for partition assignment and revocation:

```python
import asyncio
from confluent_kafka.experimental.aio import AIOConsumer

kafka_configuration_object = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-consumer-group',
    'auto.offset.reset': 'earliest'
}

topics = ['my-topic']

async def main():
    consumer = AIOConsumer(kafka_configuration_object)

    async def on_assign(consumer, partitions):
        print(f'Partitions assigned: {partitions}')
        await consumer.incremental_assign(partitions)

    async def on_revoke(consumer, partitions):
        print(f'Partitions revoked: {partitions}')
        await consumer.commit()

    async def on_lost(consumer, partitions):
        print(f'Partitions lost: {partitions}')

    await consumer.subscribe(
        topics,
        on_assign=on_assign,
        on_revoke=on_revoke,
        on_lost=on_lost
    )

    try:
        while True:
            msg = await consumer.poll(timeout=1.0)
            if msg is not None and not msg.error():
                print(f'Consumed: {msg.value()}')
                await consumer.store_offsets(message=msg)
    finally:
        await consumer.unsubscribe()
        await consumer.close()

asyncio.run(main())
```

### Synchronous writes

The Python client provides a `flush()` method which can be used to make
writes synchronous. This is typically a bad idea since it effectively
limits throughput to the broker round trip time, but may be justified in
some cases.

```python
producer.produce(topic, key="key", value="value")
producer.flush()
```

<a id="consumer-python-client"></a>

Typically, `flush()` should be called prior to shutting down the producer
to ensure all outstanding/queued/in-flight messages are delivered.

## Kafka consumer

### Initialization

The consumer is configured using a dictionary in the examples below. If you are
running Kafka locally, you can initialize the consumer as shown below.

```python
from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'host1:9092,host2:9092',
        'group.id': 'foo',
        'auto.offset.reset': 'smallest'}

consumer = Consumer(conf)
```

If you are connecting to a Kafka cluster in Confluent Cloud, you must provide
credentials for access. The example below shows using a cluster API key and
secret.

```python
from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'pkc-abcd85.us-west-2.aws.confluent.cloud:9092',
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'PLAIN',
        'sasl.username': '<CLUSTER_API_KEY>',
        'sasl.password': '<CLUSTER_API_SECRET>',
        'group.id': 'foo',
        'auto.offset.reset': 'smallest'}

consumer = Consumer(conf)
```

The `group.id` property is mandatory and specifies which consumer group the consumer
is a member of. The `auto.offset.reset` property specifies what offset the consumer
should start reading from in the event there are no committed offsets for a partition,
or the committed offset is invalid (perhaps due to log truncation).

The local example below shows `enable.auto.commit` configured to `false`
in the consumer. The default value is `True`.

```python
from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'host1:9092,host2:9092',
        'group.id': 'foo',
        'enable.auto.commit': 'false',
        'auto.offset.reset': 'earliest'}

consumer = Consumer(conf)
```

* For information on the available configuration properties, see the
  [API Documentation](/platform/current/clients/confluent-kafka-python/html/index.html).
* For a step-by-step tutorial using the Python client including code samples for
  the producer and consumer see [this guide](https://developer.confluent.io/get-started/python/).

### Python Client code examples

#### Basic poll loop

A typical Kafka consumer application is centered around a consume loop, which repeatedly calls
the `poll` method to retrieve records one-by-one that have been efficiently pre-fetched by
the consumer in behind the scenes. Before entering the consume loop, you’ll typically use the
`subscribe` method to specify which topics should be fetched from:

```python
running = True

def basic_consume_loop(consumer, topics):
    try:
        consumer.subscribe(topics)

        while running:
            msg = consumer.poll(timeout=1.0)
            if msg is None: continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    sys.stderr.write('%% %s [%d] reached end at offset %d\n' %
                                     (msg.topic(), msg.partition(), msg.offset()))
                elif msg.error():
                    raise KafkaException(msg.error())
            else:
                msg_process(msg)
    finally:
        # Close down consumer to commit final offsets.
        consumer.close()

def shutdown():
    running = False
```

The poll timeout is hard-coded to one second. If no records
are received before this timeout expires, then `Consumer.poll()` returns
an empty record set.

Note that you should always call `Consumer.close()` after you are finished
using the consumer. Doing so ensures that active sockets are
closed and internal state is cleaned up. It also triggers a group
rebalance immediately which ensures that any partitions owned by the
consumer are re-assigned to another member in the group. If not closed
properly, the broker triggers the rebalance only after the session
timeout has expired.

#### Synchronous commits

The simplest and most reliable way to manually commit offsets is by
setting the `asynchronous` parameter to the `Consumer.commit()`
method call. This method can also accept the mutually exclusive keyword
parameters `offsets` to explicitly list the offsets for each assigned
topic partition and `message` which commits offsets relative to a
`Message` object returned by `poll()`.

```python
def consume_loop(consumer, topics):
    try:
        consumer.subscribe(topics)

        msg_count = 0
        while running:
            msg = consumer.poll(timeout=1.0)
            if msg is None: continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    sys.stderr.write('%% %s [%d] reached end at offset %d\n' %
                                     (msg.topic(), msg.partition(), msg.offset()))
                elif msg.error():
                    raise KafkaException(msg.error())
            else:
                msg_process(msg)
                msg_count += 1
                if msg_count % MIN_COMMIT_COUNT == 0:
                    consumer.commit(asynchronous=False)
    finally:
        # Close down consumer to commit final offsets.
        consumer.close()
```

In this example, a synchronous commit is triggered every `MIN_COMMIT_COUNT`
messages. The `asynchronous` flag controls whether this call is
asynchronous. You could also trigger the commit on expiration of a
timeout to ensure there the committed position is updated regularly.

#### Delivery guarantees

##### Default guarantee

The following are the default settings of the Python client, and the settings
result in the default delivery guarantee of the Python client being “None”:

```text
'enable.auto.commit': 'true'
'enable.auto.offset.store': 'true'
```

Since auto commits are performed in a background thread, these settings may result in the
offset for the latest message being committed before the application has
finished processing the message. If the application were to crash or exit before
finishing processing, and the offset had been auto-committed, the next
incarnation of the consumer application would start at the next message,
effectively missing the message that was processed when the application crashed.
You can lose data or get duplicates.

##### “At least once” guarantee

To achieve “at least once” guarantee, configure the following settings:

```text
'enable.auto.commit': 'true'
'enable.auto.offset.store': 'false'
```

To avoid the scenario of data loss or duplicates with the above default “None
guaranteed” mode, the application can disable the automatic offset store and
manually store offsets (with `rd_kafka_offsets_store()`) after processing.
This gives the application fine-grained control over when a message is
committed. The latest stored offset is automatically committed every
`auto.commit.interval.ms`.

For this guarantee option, you should store the offset only after processing the
message successfully.

Note: Only offsets greater than the current offset are committed. For example, if the latest committed
offset was `10` and the application performs an `offsets_store()` with
offset `9`, that offset is not committed.

In the example in the previous section, you get “at least once” delivery since
the commit follows the message processing. By changing the order and committing
synchronously before processing, you can get “at most once” delivery, but you
must handle commit failures carefully.

```python
def consume_loop(consumer, topics):
    try:
        consumer.subscribe(topics)

        while running:
            msg = consumer.poll(timeout=1.0)
            if msg is None: continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    sys.stderr.write('%% %s [%d] reached end at offset %d\n' %
                                     (msg.topic(), msg.partition(), msg.offset()))
                elif msg.error():
                    raise KafkaException(msg.error())
            else:
                consumer.commit(asynchronous=False)
                msg_process(msg)

    finally:
        # Close down consumer to commit final offsets.
        consumer.close()
```

For simplicity in this example, `Consumer.commit()` is used
prior to processing the message. Committing on every message would
produce a lot of overhead in practice. A better approach would be to
collect a batch of messages, execute the synchronous commit, and then
process the messages only if the commit succeeded.

#### Asynchronous Commits

```python
def consume_loop(consumer, topics):
    try:
        consumer.subscribe(topics)

        msg_count = 0
        while running:
            msg = consumer.poll(timeout=1.0)
            if msg is None: continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    sys.stderr.write('%% %s [%d] reached end at offset %d\n' %
                                     (msg.topic(), msg.partition(), msg.offset()))
                elif msg.error():
                    raise KafkaException(msg.error())
            else:
                msg_process(msg)
                msg_count += 1
                if msg_count % MIN_COMMIT_COUNT == 0:
                    consumer.commit(asynchronous=True)
    finally:
        # Close down consumer to commit final offsets.
        consumer.close()
```

In this example, the consumer sends the request and returns
immediately by using asynchronous commits. The `asynchronous` parameter to `commit()` is
changed to `True`. The value is passed in explicitly, but asynchronous
commits are the default if the parameter is not included.

The API gives you a callback which is invoked
when the commit either succeeds or fails.
The commit callback can be any callable and can be passed
as a configuration parameter to the consumer constructor.

```python
from confluent_kafka import Consumer

def commit_completed(err, partitions):
    if err:
        print(str(err))
    else:
        print("Committed partition offsets: " + str(partitions))

conf = {'bootstrap.servers': "host1:9092,host2:9092",
        'group.id': "foo",
        'default.topic.config': {'auto.offset.reset': 'smallest'},
        'on_commit': commit_completed}

consumer = Consumer(conf)
```

## Kafka share consumers

A share consumer reads from a share group where multiple consumers
cooperatively consume the same partitions. The broker tracks progress
for each individual record rather than by committed offset. This lets you
scale the number of consumers beyond the number of partitions and
distribute work like a traditional queue.

Share consumers use a separate `ShareConsumer` class, not the regular `Consumer`
class. The broker drives partition assignment, so there is no rebalance callback and no
`assign()` step.

For more about share consumers, see the documentation:

* [Confluent Platform documentation](https://docs.confluent.io/platform/current/clients/share-consumers.html)
* [Confluent Cloud documentation](https://docs.confluent.io/cloud/current/client-apps/share-consumers.html)
* [Share Consumer (Queues for Kafka) API documentation](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#kip-932-share-consumer-queues-for-kafka)
* [ShareConsumer(config) API documentation](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.ShareConsumer.ShareConsumer)

#### NOTE
Share consumers are a Preview feature. To use share consumers,
your Kafka cluster must have share groups enabled, which are available in Confluent Cloud
and Confluent Platform 8.2 and later.

A Preview feature is a Confluent component that is introduced to gain
early feedback from developers. You can use Preview features for evaluation
and non-production testing purposes or to provide feedback to Confluent.
The warranty, SLA, and Support Services provisions of your agreement with
Confluent don’t apply to Preview features. Confluent might discontinue
providing preview releases of the Preview features at any time in
Confluent’s sole discretion.

### Basic poll loop (implicit acknowledgement)

By default the consumer is in implicit acknowledgement mode: you don’t
acknowledge records yourself. Every record returned by a poll is
automatically accepted on the next call to `poll()`,
`commit_sync()`, or `commit_async()`. For normal consumers, `poll()` returns a
single record. For share consumers, `poll()` returns a batch
of records that might be empty rather than a single message, so always
iterate the result. Implicit mode is at-least-once, so design your
processing to tolerate occasional redelivery.

This example shows how to import a share consumer. For complete examples,
see the [Python Client repo](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples).

```python
from confluent_kafka import ShareConsumer

consumer = ShareConsumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-share-group',
})
consumer.subscribe(['my-topic'])

try:
    while running:
        messages = consumer.poll(timeout=1.0)  # a batch, possibly empty
        for msg in messages:
            if msg.error():
                continue
            process(msg)  # auto-accepted on the next poll
finally:
    consumer.close()
```

### Explicit acknowledgement

Set `share.acknowledgement.mode` to `explicit` to acknowledge every record
returned by a poll yourself, before the next poll. Acknowledge a record with
`ACCEPT` when processed, `RELEASE` for transient failures to make it
available again for a later delivery attempt, or `REJECT` for permanent
failures, such as a poison record, to stop redelivering it. This includes
records delivered with an error, which you acknowledge, typically with
`REJECT`, so that every record in the batch is acknowledged before the next
`poll()`. Flush
acknowledgements to the broker with `commit_sync()`, which blocks and returns
a per-partition result, or with `commit_async()`.

```python
from confluent_kafka import ShareConsumer, AcknowledgeType

consumer = ShareConsumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-share-group',
    'share.acknowledgement.mode': 'explicit',
})
consumer.subscribe(['my-topic'])

try:
    while running:
        messages = consumer.poll(timeout=1.0)
        for msg in messages:
            if msg.error():
                consumer.acknowledge(msg, AcknowledgeType.REJECT)
                continue
            try:
                process(msg)
                consumer.acknowledge(msg, AcknowledgeType.ACCEPT)
            except TransientError:
                consumer.acknowledge(msg, AcknowledgeType.RELEASE)
            except Exception:
                consumer.acknowledge(msg, AcknowledgeType.REJECT)
        # Flush acknowledgements. A None value means that partition succeeded.
        results = consumer.commit_sync(timeout=10.0)
        for tp, exc in results.items():
            if exc is not None:
                print(f"commit failed for {tp.topic} [{tp.partition}]: {exc}")
finally:
    consumer.close()
```

<a id="fips-compliance"></a>

## FIPS compliance

This client supports both FIPS 140-2 and FIPS 140-3 compliance. Use the following version mapping to ensure compliance:

| Compliance Standard   | OpenSSL Version   | FIPS Provider Version                                                         |
|-----------------------|-------------------|-------------------------------------------------------------------------------|
| FIPS 140-2            | 3.x               | [3.0.8](https://github.com/openssl/openssl/blob/openssl-3.0.8/README-FIPS.md) |
| FIPS 140-3            | 3.x               | [3.1.2](https://github.com/openssl/openssl/blob/openssl-3.1.2/README-FIPS.md) |

For new deployments, use FIPS 140-3 because federal procurement will no longer accept FIPS 140-2 certificates issued after September 21, 2026.

Confluent tested communication for FIPS compliance between clients and the following endpoints:

- Kafka brokers
- Schema Registry

### Kafka broker and Schema Registry

Kafka broker
: To communicate with the Kafka broker, the client uses the librdkafka library, which uses OpenSSL.
  The steps below configure OpenSSL to operate in FIPS mode for client communication with the broker.

Schema Registry
: To communicate with Schema Registry, the client uses the standard libraries in Python, which uses the operating
  system’s native SSL/TLS library. For FIPS-compliant communication between Schema Registry and the client, do
  not use the steps that follow. Instead, make the SSL/TLS library FIPS compliant. If the native
  SSL/TLS library is OpenSSL (the default for Python), then use the steps in the
  [OpenSSL readme](https://github.com/openssl/openssl/blob/openssl-3.0.8/README-FIPS.md) to make OpenSSL
  FIPS compliant.

### FIPS-compliant Kafka broker communication

For FIPS-compliant communication with the broker, there are two ways to approach client installation:

- Use prebuilt wheels
- Build librdkafka and the client both from source

#### Use prebuilt wheels

If you install this client through prebuilt wheels using `pip install confluent_kafka`, OpenSSL 3.0
is already statically linked with the librdkafka shared library. To enable this client to
communicate with the Kafka cluster using the OpenSSL FIPS provider and FIPS-approved algorithms,
you must enable the FIPS provider. You can find steps to enable the FIPS provider in section
[Use FIPS provider](#enabling-fips-provider).

#### NOTE
You should enable the FIPS provider (using the same steps) if you install this client from the source using
`pip install confluent_kafka --no-binary :all:` with prebuilt librdkafka in which OpenSSL is statically linked.

#### Build librdkafka and client both from source

When you build the librdkafka from source, librdkafka dynamically links to the OpenSSL present in the system
if static linking is not used explicitly while building. If the system installed OpenSSL is already working in FIPS mode,
then you can directly jump to the section
[client configuration to enable FIPS provider](#client-configuration-to-enable-fips-provider) and enable the `fips` provider.

If you don’t have OpenSSL working in FIPS mode, use the steps mentioned in the section
[Use FIPS provider](#enabling-fips-provider) to make OpenSSL in your system FIPS compliant, and then enable the `fips`
provider. After you have OpenSSL working in FIPS mode and the `fips` provider enabled, librdkafka and the Python Client
will use FIPS approved algorithms for the communication between client and Kafka cluster.

<a id="enabling-fips-provider"></a>

### Use FIPS provider

To use the FIPS provider, you must have the FIPS module available on your system. Plug the module into OpenSSL,
and then configure OpenSSL to use the module.

You can plug the FIPS provider into OpenSSL two ways:

- Put the module in the default module folder of OpenSSL.
- Point to the module with the environment variable, `OPENSSL_MODULES`. For
  example: `OPENSSL_MODULES="/path/to/fips/module/lib/folder/`.

After you plug the FIPS provider module into OpenSSL, you must configure OpenSSL to use the module. Once again, you have two options:

- Modify the default configuration file to include the FIPS-related config.
- Create a new configuration file and point to it using the environment variable, `OPENSSL_CONF`.
  For example: `OPENSSL_CONF="/path/to/fips/enabled/openssl/config/openssl.cnf`.

  For an example of OpenSSL configuration file, see: [Enable FIPS provider with OpenSSL](#link-fips-provider-with-openssl).

#### NOTE
You must specify both `OPENSSL_MODULES` and `OPENSSL_CONF` environment variables when installing the client
from pre-built wheels or when OpenSSL is statically linked to librdkafka.

#### Build FIPS provider module

This section provides a high-level overview of how to build the FIPS provider module. To find the official steps to generate
the FIPS provider module, see the following version-specific documentation:

- [For FIPS 140-2 (v3.0.8)](https://github.com/openssl/openssl/blob/openssl-3.0.8/README-FIPS.md).
- [For FIPS 140-3 (v3.1.2)](https://github.com/openssl/openssl/blob/openssl-3.1.2/README-FIPS.md).

To build the FIPS provider module:

1. Clone OpenSSL from: [OpenSSL Github Repo](https://github.com/openssl/openssl).
2. Use `git checkout` to checkout a FIPS-compliant version of OpenSSL 3.0. The latest version may not be FIPS-compliant. At the time of this writing,
   v3.1.2 (tagged as [v3.1.2](https://github.com/openssl/openssl/tree/openssl-3.1.2)) and v3.0.8 (tagged as [v3.0.8](https://github.com/openssl/openssl/tree/openssl-3.0.8)) are currently tested for FIPS 140-3 and FIPS 140-2 respectively.
3. Run: `./Configure enable-fips`.
4. Run: `make install_fips`.

   Inside the `providers` folder, two files are generated. Use these files with OpenSSL:
   - FIPS module (fips.dylib in Mac, fips.so in Linux, and fips.dll in Windows)
   - FIPS config (fipsmodule.cnf)

#### Reference FIPS provider in OpenSSL

When installing from source, you can dynamically plug the FIPS module built above into OpenSSL by putting the FIPS module into the
default OpenSSL module folder. Look for something like: `...lib/ossl-modules/`.

For the default locations of OpenSSL on various operating
systems, see the SSL section of the [Introduction to librdkafka - the Apache Kafka C/C++ client library](https://github.com/confluentinc/librdkafka/blob/master/INTRODUCTION.md#ssl).

You can also point to this module with the environment variable `OPENSSL_MODULES`.

For example: `OPENSSL_MODULES="/path/to/fips/module/lib/folder/`.

<a id="link-fips-provider-with-openssl"></a>

#### Enable FIPS provider with OpenSSL

To enable FIPS in OpenSSL, you must include `fipsmodule.cnf` in the file, `openssl.cnf`. See the
following `openssl.cnf` example:

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

The `fipsmodule.cnf` file includes `fips_sect` which OpenSSL requires to enable FIPS.

Some of the algorithms might have different implementation in FIPS or other providers. If you load two different
providers like `default` and `fips`, any implementation could be used. To make sure you fetch only FIPS-compliant version
of the algorithm, use `fips=yes` default property in config file.

<a id="client-configuration-to-enable-fips-provider"></a>

#### Client configuration to enable FIPS provider

OpenSSL requires some non-crypto algorithms as well. These algorithms are not included in the FIPS
provider and you must use the `base` provider in conjunction with the `fips` provider. Base
provider comes with OpenSSL by default. You must enable `base` provider in the client configuration.

To make the client (consumer, producer or admin client) FIPS compliant, you must enable
only `fips` and `base` provider in the client using the `ssl.providers` configuration property.

Configure the property this way: `'ssl.providers': 'fips,base'`.

<a id="asyncio-python-client"></a>

## AsyncIO support

The Python client provides AsyncIO-compatible producer and consumer clients for
integration with async Python applications. Use the AsyncIO clients when your
application uses Python’s `asyncio` framework and you require non-blocking Kafka
operations that integrate seamlessly with other async operations.

#### NOTE
The AsyncIO API is available under the `experimental` package
`confluent_kafka.experimental.aio`. The API may be subject to change
in future releases. You can use this `pip` command to install the
experimental classes:

```none
pip install --pre confluent-kafka
```

### When to use AsyncIO clients

Choose the AsyncIO or synchronous client based on your application architecture.

Use AsyncIO clients when:

* Your application runs under an event loop (FastAPI, Starlette, aiohttp, Sanic,
  asyncio workers).
* You must avoid blocking the event loop during Kafka operations.
* You require integration Kafka with other async I/O operations.
* You’re building async web services or microservices.

Use synchronous clients when:

* Building scripts, batch jobs, or CLI tools.
* Running high-throughput data pipelines where you control threads and processes.
* You are able to call `poll()` and `flush()` directly without blocking issues.
* You require per-message headers in produce operations (not supported in async
  batched path).

In an asynchronous server environment, prefer using the AsyncIO client for
better compatibility. If you require headers, invoke the synchronous `produce()` using
`run_in_executor()` for that path.

<a id="asyncio-producer-python-client"></a>

### AsyncIO producer

#### Initialize the producer

Import `AIOProducer` from `confluent_kafka.experimental.aio` and configure it using a
dictionary, similar to the synchronous `Producer`.

### Confluent Cloud

For Confluent Cloud connections, provide credentials in the configuration dictionary:

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

async def produce_to_ccloud():
    conf = {
        'bootstrap.servers': 'pkc-abcd85.us-west-2.aws.confluent.cloud:9092',
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'PLAIN',
        'sasl.username': '<CLUSTER_API_KEY>',
        'sasl.password': '<CLUSTER_API_SECRET>',
        'client.id': 'my-async-producer'
    }
    producer = AIOProducer(conf)

    try:
        # Your async produce logic here
        pass
    finally:
        await producer.close()

asyncio.run(produce_to_ccloud())
```

### Confluent Platform

For Confluent Platform connections, configure the `AIOProducer` as follows.
This example assumes a local, unsecured cluster.

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

async def produce_example():
    producer = AIOProducer({
        'bootstrap.servers': 'localhost:9092',
        'client.id': 'my-async-producer'
    })

    try:
        # Your async produce logic here
        pass
    finally:
        await producer.close()

asyncio.run(produce_example())
```

#### Produce messages asynchronously

Call the `produce` method with `await` to enqueue messages for delivery. The
method returns a `Future` that resolves to the delivered message.

```python
async def send_message(producer, topic):
    # Produce returns a Future
    delivery_future = await producer.produce(
        topic=topic,
        key='key1',
        value='value1'
    )

    # Await the future to get delivery confirmation
    msg = await delivery_future
    print(f'Delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

asyncio.run(send_message())
```

#### Produce messages in batches

Produce multiple messages concurrently and await their delivery using
`asyncio.gather()`:

```python
import asyncio
async def batch_produce(producer, topic):
    # Create multiple produce futures
    futures = [
        await producer.produce(topic=topic, key=f'key{i}', value=f'value{i}')
        for i in range(100)
    ]

    # Flush to ensure messages are in flight
    await producer.flush()

    # Wait for all deliveries
    messages = await asyncio.gather(*futures)
    print(f'Delivered {len(messages)} messages')

asyncio.run(def batch_produce())
```

#### Produce messages transactionally

The `AIOProducer` supports async transactional operations:

```python
import asyncio
from confluent_kafka.experimental.aio import AIOProducer
async def transactional_produce(topic):
    producer = AIOProducer({
        'bootstrap.servers': 'host1:9092,host2:9092',
        'transactional.id': 'my-transactional-producer'
    })

    await producer.init_transactions()

    try:
        await producer.begin_transaction()

        # Produce messages within transaction
        futures = [
            await producer.produce(topic=topic, value=f'msg{i}')
            for i in range(10)
        ]
        await producer.flush()
        await asyncio.gather(*futures)

        # Commit the transaction
        await producer.commit_transaction()
    except Exception as e:
        # Abort on error
        await producer.abort_transaction()
        raise
    finally:
        await producer.close()

asyncio.run(transactional_produce())
```

#### Close the producer

<a id="asyncio-consumer-python-client"></a>

### AsyncIO consumer

#### Initialize the consumer

Import `AIOConsumer` from `confluent_kafka.experimental.aio` and configure it
using a dictionary.

### Confluent Cloud

For Confluent Cloud connections, provide credentials in the configuration dictionary:

```python
import asyncio
from confluent_kafka.experimental.aio import AIOConsumer

async def consume_from_ccloud():
    conf = {
        'bootstrap.servers': 'pkc-abcd85.us-west-2.aws.confluent.cloud:9092',
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'PLAIN',
        'sasl.username': '<CLUSTER_API_KEY>',
        'sasl.password': '<CLUSTER_API_SECRET>',
        'group.id': 'my-consumer-group',
        'auto.offset.reset': 'earliest'
    }
    consumer = AIOConsumer(conf)

    try:
        # Your async consume logic here
        pass
    finally:
        await consumer.close()

asyncio.run(consume_from_ccloud())
```

### Confluent Platform

For Confluent Platform connections, configure the `AIOConsumer` as follows.
This example assumes a local, unsecured cluster.

```python
import asyncio
from confluent_kafka.experimental.aio import AIOConsumer

async def consume_example():
    consumer = AIOConsumer({
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'my-consumer-group',
        'auto.offset.reset': 'earliest'
    })

    try:
        # Your async consume logic here
        pass
    finally:
        await consumer.close()

asyncio.run(consume_example())
```

#### Consume messages in a loop

Use `await` with the `poll()` method to consume messages without blocking
the event loop:

```python
import asyncio
from confluent_kafka.experimental.aio import AIOConsumer

# Use the same configuration keys shown in the initialization section
kafka_configuration_object = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-consumer-group',
    'auto.offset.reset': 'earliest'
}

topics = ['my-topic']

async def main():
    consumer = AIOConsumer(kafka_configuration_object)
    await consumer.subscribe(topics)

    try:
        while True:
            # Poll without blocking the event loop
            msg = await consumer.poll(timeout=1.0)

            if msg is None:
                continue

            if msg.error():
                print(f'Consumer error: {msg.error()}')
                continue

            # Process the message
            print(f'Received: {msg.value().decode("utf-8")}')
    finally:
        await consumer.unsubscribe()
        await consumer.close()

asyncio.run(main())
```

#### Manage offsets manually

Disable `auto.commit` and manually commit offsets for precise control:

```python
import asyncio
from confluent_kafka.experimental.aio import AIOConsumer

kafka_configuration_object = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-consumer-group',
    'enable.auto.commit': 'false',
    'enable.auto.offset.store': 'false',
    'auto.offset.reset': 'earliest'
}

topics = ['my-topic']

async def main():
    consumer = AIOConsumer(kafka_configuration_object)
    await consumer.subscribe(topics)

    msg_count = 0
    try:
        while True:
            msg = await consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                continue

            # Process the message
            print(f'Received: {msg.value()}')

            # Store offset after processing
            await consumer.store_offsets(message=msg)

            msg_count += 1
            # Commit every 100 messages
            if msg_count % 100 == 0:
                await consumer.commit()
    finally:
        await consumer.unsubscribe()
        await consumer.close()

asyncio.run(main())
```

#### Integrate with Schema Registry

The async Schema Registry client and its associated serializers are
experimental and their APIs, including the initialization pattern, may change
in future versions. They are accessed via private modules (e.g., `_async`)
to indicate their unstable status.

#### Produce Avro messages asynchronously

```python
from confluent_kafka.schema_registry import AsyncSchemaRegistryClient
from confluent_kafka.schema_registry._async.avro import AsyncAvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.experimental.aio import AIOProducer

async def produce_avro_async():
    # Configure async Schema Registry client
    sr_client = AsyncSchemaRegistryClient({
        'url': 'http://localhost:8081'
    })

    # Define Avro schema...
    schema_str = '''
    {
        "type": "record",
        "name": "User",
        "fields": [{"name": "name", "type": "string"}]
    }
    '''

    # Await the serializer constructor to complete async initialization
    avro_serializer = await AsyncAvroSerializer(
        sr_client,
        schema_str=schema_str
    )

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

    try:
        # Serialize and produce
        value = {'name': 'alice'}
        serialized = await avro_serializer(
            value,
            SerializationContext('my-topic', MessageField.VALUE)
        )
        delivery_future = await producer.produce(
            'my-topic',
            value=serialized
        )
        msg = await delivery_future
        print(f'Delivered to {msg.topic()}')
    finally:
        await producer.flush()
        await producer.close()

asyncio.run(produce_avro_async())
```

#### Consume Avro messages asynchronously

```python
from confluent_kafka.schema_registry import AsyncSchemaRegistryClient
from confluent_kafka.schema_registry._async.avro import AsyncAvroDeserializer
from confluent_kafka.experimental.aio import AIOConsumer
from confluent_kafka.serialization import SerializationContext, MessageField

async def consume_avro_async():
    # Configure async Schema Registry client
    sr_client = AsyncSchemaRegistryClient({
        'url': 'http://localhost:8081'
    })

    # Await the deserializer constructor to complete async initialization
    avro_deserializer = await AsyncAvroDeserializer(sr_client)

    consumer = AIOConsumer({
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'my-avro-consumer',
        'auto.offset.reset': 'earliest'
    })

    await consumer.subscribe(['my-topic'])

    try:
        msg = await consumer.poll(timeout=5.0)
        if msg is not None and not msg.error():
            value = await avro_deserializer(
                msg.value(),
                SerializationContext('my-topic', MessageField.VALUE)
            )
            print(f'Received Avro value: {value}')
    finally:
        await consumer.unsubscribe()
        await consumer.close()

asyncio.run(consume_avro_async())
```

## Examples

### Producer and consumer examples

The [basic producer and consumer examples](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md#basic-producerconsumer-examples)
work with any Kafka deployment, but are optimized for Confluent Cloud and Confluent Platform which provide additional features, security, and enterprise support.

- [producer.py](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/producer.py) Read lines from stdin and send them to a Kafka topic.
- [consumer.py](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/consumer.py) Read messages from a Kafka topic.
- [context_manager_example.py](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/context_manager_example.py) Demonstrates
  context manager `with` statement usage for the producer, consumer, and AdminClient, including automatic resource
  cleanup when exiting the `with` block.

### AsyncIO examples

The [AsyncIO examples](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/README.md#python-client-examples-for-apache-kafka)
examples demonstrate AsyncIO patterns including concurrent producer and
consumer operations, signal handling, and transaction management.

- [asyncio_example.py](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/asyncio_example.py)
  Comprehensive AsyncIO example that demonstrates both AIOProducer and AIOConsumer
  with transactional operations, batched async produce, proper event loop integration, signal handling,
  and async callback patterns.
- [asyncio_avro_producer.py](https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/asyncio_avro_producer.py)
  Minimal AsyncIO Avro producer using AsyncSchemaRegistryClient and AsyncAvroSerializer. Supports Confluent Cloud
  using `--sr-api-key/--sr-api-secret`.

## API documentation

To view the Python client API documentation, click [here](/platform/current/clients/confluent-kafka-python/html/index.html) .

## Related content

* [Apache Kafka for Python Developers](https://developer.confluent.io/learn-kafka/kafka-python/intro/) (Confluent Developer)
* [Apache Kafka 101](https://developer.confluent.io/learn-kafka/apache-kafka/)  (Confluent Developer)
* [Confluent Developer Python tutorial](https://developer.confluent.io/get-started/python/)  (Confluent Developer)
* [Getting Started with Apache Kafka and Python](https://www.confluent.io/blog/getting-started-with-apache-kafka-in-python/) (Confluent Blog)
* [Integrating Apache Kafka With Python Asyncio Web Applications](https://www.confluent.io/blog/kafka-python-asyncio-integration/) (Confluent Blog)
* [Generating the FIPS module and config file](https://github.com/openssl/openssl/blob/openssl-3.0.8/README-FIPS.md) (GitHub)
* [How to use the FIPS Module](https://www.openssl.org/docs/man3.0/man7/fips_module.html) (OpenSSL Documentation)
* [librdkafka SSL Information](https://github.com/confluentinc/librdkafka/blob/master/INTRODUCTION.md#ssl) (GitHub)