Metadata-Version: 2.4
Name: confluent-kafka
Version: 4.4.0b1.dev4
Summary: Python bindings for the Confluent Kafka Rust client
Author: Confluent Inc.
License: Apache-2.0
Project-URL: Repository, https://github.com/confluentinc/kafka-clients
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Provides-Extra: dev
Requires-Dist: psutil; extra == "dev"
Requires-Dist: matplotlib; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: testcontainers>=4.0; extra == "dev"
Requires-Dist: typing_extensions>=4.2; extra == "dev"
Provides-Extra: docs
Requires-Dist: sphinx>=8.1; extra == "docs"
Requires-Dist: furo>=2025.12.19; extra == "docs"

# confluent_kafka (Rust-core Python client)

The Python client built on the Rust core's C FFI. Its surface is the Java Kafka
client (Apache Kafka 4.4.0) mapped to idiomatic Python, ruled by the
`## Python Binding Conventions` section of the repository's `CLAUDE.md`. The
package sits over a hand-written CPython extension, `_confluentkafka`, which
links the Rust library's C API (`../rust`, feature `ffi`).

## Status

**Preview:** the Python client is in preview and not yet recommended for production use. The public API is not stable before the GA release and may change. We welcome your feedback while the design is still open.

Windows is not supported yet.

**Known issue on Python 3.14:** a Ctrl+C (`KeyboardInterrupt`) can, rarely, leave a client call counted, so that a later `close()` waits forever, or leave a mock client's lock held. It happens when the Ctrl+C arrives right after a `with` block takes its lock and before the block starts. The bug is in CPython 3.14 and affects every `with` block ([python/cpython#148874](https://github.com/python/cpython/issues/148874)); Python 3.15 has the fix.

## Import paths

The package mirrors Java's package tree with the `clients` segment dropped
(`CLAUDE.md`, Python Binding Conventions, Modules). Import each name from its own
module — the package root does **not** re-export the everyday client names; it
exports only `Duration` and the Java built-in exception classes:

```python
from confluent_kafka.producer import (
    KafkaProducer, MockProducer, AsyncKafkaProducer, AsyncMockProducer,
    ProducerRecord, RecordMetadata, Callback,
)
from confluent_kafka.admin import Admin, AsyncAdmin, NewTopic
from confluent_kafka.consumer import (
    KafkaConsumer, MockConsumer, AsyncKafkaConsumer, AsyncMockConsumer,
    ConsumerRebalanceListener, OffsetAndMetadata, CloseOptions,
    SubscriptionPattern,
)
from confluent_kafka.common import (
    TopicPartition, TopicIdPartition, Node, PartitionInfo, Uuid,
    MetricName, Metric,
    KafkaError,  # Java's org.apache.kafka.common.KafkaException
    KafkaFuture, AsyncKafkaFuture,  # every future the package hands out
)
# Every type lives in the module of its Java package, `clients` dropped:
from confluent_kafka.common.header import Header, Headers  # common.header
from confluent_kafka.common.metrics import KafkaMetric  # common.metrics
from confluent_kafka.common.record import TimestampType  # common.record
from confluent_kafka.common.errors import RetriableError  # typed hierarchy
from confluent_kafka.common.serialization import (
    bytes_serializer, string_serializer, int_serializer, json_serializer,  # etc.
)

# The Java built-in exception classes and the Duration alias live at the
# package root:
from confluent_kafka import (
    IllegalArgumentError, IllegalStateError, ConcurrentModificationError,
    TimeoutError, NoSuchElementError, ExecutionError, UnsupportedOperationError,
    IndexOutOfBoundsError, Duration,
)
```

The native extension is `_confluentkafka` (private): the package imports it; it
is never a public import path.

## `None` is Java's `null`

Every keyword-only parameter mirrors a Java argument, and `None` is Java's
`null`, never "not given": leaving a parameter out is the only way to choose the
Java overload without it (`CLAUDE.md`, Python Binding Conventions, Signatures).
`consumer.close()` is Java's `close()`, while `consumer.close(option=None)` is
`close(null)` and raises `TypeError("option must not be None")`, as Java's
`NullPointerException`; `producer.send(record=r, callback=None)` is Java's
`send(record, null)`. Where Java gives `null` a meaning, `None` keeps it
(`ProducerRecord(topic=…, partition=None, key=None, value=…)`, an
`Optional` parameter's `Optional.empty()`). Code written against kafka-python or
confluent-kafka-python that passes `None` to skip an argument must leave the
argument out instead; its positional callbacks and serdes take the keyword form
(Serdes and callbacks), and an `except UnsupportedVersionError` meant for a mock's
unimplemented method becomes `except UnsupportedOperationError` (Errors). The
signatures show the private default `MISSING`
(`help()`, `inspect.signature`): it only marks a parameter some overload omits
and is never a value to pass.

## Headers

A record's headers are Java's: `record.headers()` returns a `Headers`
(`confluent_kafka.common.header`), never `None` and empty when there are none, holding
`Header` objects with `key()` and `value()`, a `memoryview` or `None` (for a
fetched record, a view into the fetch batch, never a copy). `Headers` has
Java's `add(header=…)` / `add(key=…, value=…)`, `remove(key=…)`,
`last_header(key=…)`, `headers(key=…)` and `to_array()`, and iterates with
`for h in record.headers()`. A record's headers can be changed until it is sent
(by `KafkaProducer.send`, once the serializers ran); then `add` and `remove`
raise `IllegalStateError("RecordHeaders has been closed.")`, as Java's
`KafkaProducer` seals them. A polled record's headers stay writable, as Java's
`CompletedFetch` leaves them. Where headers are taken (`ProducerRecord(headers=…)`,
`ConsumerRecord(headers=…)`), each element is a `Header(key=…, value=…)` or a
`(key, value)` pair. Code that read headers as `(key, value)` tuples
(`for k, v in record.headers()`) reads `h.key()`, `h.value()` instead.

## Async peers

Each async client lives in the same module as its sync peer, `Async`-prefixed
(`AsyncKafkaProducer`, `AsyncKafkaConsumer`). A method is `async def` on the
async class iff Java waits in it (on the background thread, the network, or a
rebalance listener it runs); the others (`assignment()`, `subscription()`,
`metrics()`, `group_metadata()`, `wakeup()`) stay plain `def` on both.
`commit_nowait()` is `async def` on the async consumers, as every method whose
entry point has a `_cb` twin: `_nowait` means the call does not wait for the
broker's answer, so `await consumer.commit_nowait(callback=cb)` returns once the
commit is sent (raising the call's own failure) and `cb` gets the result later.

## Futures

One future type for every future the package hands out: `KafkaFuture[T]` on a
sync client, `AsyncKafkaFuture[T]` on an async one (both in
`confluent_kafka.common`), with Java's `KafkaFuture` methods (`get`, `get_now`,
`then_apply`, `when_complete`, `all_of`, `cancel`, `is_done`, …).

```python
md = producer.send(record=r).get()                 # sync
md = await (await producer.send(record=r)).get()   # async: send() and get() are awaited
```

`get()` on a failed future raises `confluent_kafka.ExecutionError`, as Java's
`ExecutionException`, and the real error is its `__cause__`:

```python
try:
    producer.send(record=r).get()
except ExecutionError as e:
    if isinstance(e.__cause__, RetriableError): ...
```

A `get(timeout=…)` that runs out raises `TimeoutError`. A future's
`then_apply` functions and `when_complete` actions run where the completion
runs: on the producer's completion thread or the sync admin client's pump
thread, on the event loop for an async client.

## Serdes and callbacks take keyword arguments

A serializer, a deserializer, a `send()` callback and a `commit_nowait()`
callback are called with keyword arguments named after Java's parameters, and
each must accept `**kwargs`:

```python
def on_done(*, metadata, exception, **kwargs): ...          # Java's Callback
def on_commit(*, offsets, exception, **kwargs): ...         # OffsetCommitCallback
def serializer(*, topic, data, headers=None, **kwargs): ...     # Serializer
def deserializer(*, topic, data, headers=None, **kwargs): ...   # Deserializer

producer.send(record=r, callback=lambda *, metadata, exception, **_: ...)
```

Java adds a parameter to such an interface as a `default` overload, which old
implementations never see; in Python the binding passes it as one more keyword,
which `**kwargs` takes, so a parameter Java adds later breaks no existing
callable. A callable without `**kwargs` is rejected where it is given, with
`IllegalArgumentError("<param> must accept **kwargs")` (`callback`,
`key_serializer`, … or the config key `value.deserializer`, …), and the
`Callback`, `OffsetCommitCallback`, `Serializer` and `Deserializer` protocols
make `mypy` reject it too. The parameters may be declared positional-or-keyword
in Java's order (`def ser(topic, data, headers=None, **kwargs)`), which the
client then calls positionally, a little faster. The `ConsumerRebalanceListener`
methods, which you override on a subclass, are called positionally, as before.
Code written against the earlier positional callbacks
(`lambda md, exc: ...`, `def deserializer(topic, data, headers)`) must be
rewritten in this form.

## Producer

- `send(record=…, callback=cb)` returns a `KafkaFuture` for the record's
  `RecordMetadata` (`.get()` waits for it); like Java's, it cannot be cancelled,
  and a send failure reaches `get()` as `ExecutionError`. A serializer is called
  with a `None` key or value as given and decides what null maps to.
  `cb(metadata=…, exception=…)` is Java's `Callback`: it runs on the producer's background
  completion thread (`KafkaProducer`) or on the event loop
  (`AsyncKafkaProducer`), one at a time in completion order, before the future
  completes; a raising callback fails nothing, and the client's core logs its
  exception once, with Java's message. `metadata` is never `None`: on
  failure every field but the topic and partition is -1.
- `close()` waits for the records sent before it; `close(timeout=…)` waits at
  most `timeout`, then fails what is left. Neither cancels a future.
- A `send()` that returned belongs to the `flush()` or transaction call after it
  (`commit_transaction()` includes it, `abort_transaction()` discards it).
- `MockProducer` is Java's `MockProducer`: sends complete, and callbacks run, on
  the calling thread, and a raising callback's exception propagates out of the
  call that completed the record.

## Consumer

- Set `group.protocol` to `consumer`: this client implements only the KIP-848
  consumer group protocol, and Java's default, `classic`, is rejected by the
  client's core when the consumer is built, with
  `UnsupportedVersionError("Classic group protocol is not yet supported in this
  client; set group.protocol=consumer (KIP-848).")`.
- The share consumer (KIP-932: `ShareConsumer`, `KafkaShareConsumer`,
  `MockShareConsumer`) is not generated yet: the C FFI has no entry point for
  it.
- The calls that wait take Java's `Duration` overloads as an optional keyword:
  `commit(timeout=…)`, `position(…, timeout=…)`, `committed(…, timeout=…)`,
  `partitions_for(…, timeout=…)`, `list_topics(timeout=…)`,
  `offsets_for_times(…, timeout=…)`, `beginning_offsets(…, timeout=…)` and
  `end_offsets(…, timeout=…)`; left out calls the overload without it
  (`default.api.timeout.ms` applies), and `None` is Java's null `Duration`
  (`TypeError("timeout must not be None")`). A negative timeout raises
  `IllegalArgumentError`.
- `poll(timeout=…)` runs the deserializers on the caller's thread. A null key or
  value is not deserialized: the record's field is `None`. A failing one
  raises `RecordDeserializationError` (after the records before it, if any, were
  returned) and leaves the position at the record:
  `seek(partition=e.topic_partition(), offset=e.offset() + 1)` skips it.
- `commit()` is Java's `commitSync` and waits; `commit_nowait(callback=cb)` is
  Java's `commitAsync`. `cb(offsets=…, exception=…)` is Java's
  `OffsetCommitCallback`; `offsets` is `None` when the commit failed and for
  an explicit empty `offsets`, as Java passes `null`.
- The `ConsumerRebalanceListener` methods and the `commit_nowait()` callback run
  on the caller's thread, inside the call that delivers them, and the
  rebalance does not advance until the listener returns. As in Java,
  listener methods and background errors are delivered only by `poll()` and
  `unsubscribe()` (and `close()` revokes or loses the group assignment through
  the listener); the `commit_nowait()` callback runs inside a later `poll()`,
  `commit()`, `commit_nowait()` or `close()`; `commit()` and
  `commit_nowait()` deliver no listener method. A listener may call
  back into its consumer (`commit()`, `seek()`, `position()`, …); on
  `AsyncKafkaConsumer` those calls block the event loop for their duration.
  On `AsyncKafkaConsumer` a listener method may be `async def`; it is awaited
  on the event loop. A listener's `KafkaError` is raised as it is from the
  call that delivered it; any other exception is wrapped as
  `KafkaError("User rebalance callback throws an error")` with it as the
  cause.
- `KafkaConsumer` is not thread-safe: a call while another thread is inside it
  raises `ConcurrentModificationError`, `wakeup()` and `metrics()` excepted, as
  Java leaves both outside `acquire()`: from any thread, `wakeup()` makes the
  waiting call raise `WakeupError`, and `metrics()` answers.
- `close()` closes with Java's default timeout; `close(option=CloseOptions(timeout=…))`
  bounds it. Java's deprecated `close(Duration)` is not generated.
- A deserializer defaults to the `key.deserializer` / `value.deserializer`
  config key, else to `bytes_deserializer()`; the consumer closes both when it
  closes.
- `MockConsumer` is Java's `MockConsumer`: `add_record(record=…)` takes the
  `ConsumerRecord` the test builds and `poll()` returns it as is (the mock has no
  deserializers). Nothing binds its type parameters, so annotate it as Java
  writes them: `c: MockConsumer[str, str] =
  MockConsumer(offset_reset_strategy="earliest")`.

## Admin

`Admin.create(configs={...})` returns a `KafkaAdminClient` (`AsyncAdmin.create` an
`AsyncKafkaAdminClient`); the client is a context manager, and leaving the block
closes it. The package is `confluent_kafka.admin`: the client, one `…Options` per
call, one `…Result` per call, and the value types (`NewTopic`, `TopicDescription`, …).

```python
from confluent_kafka import ExecutionError
from confluent_kafka.admin import Admin, NewTopic
from confluent_kafka.common.errors import TopicExistsError

with Admin.create(configs={"bootstrap.servers": "localhost:9092"}) as admin:
    result = admin.create_topics(
        new_topics=[NewTopic(name="demo", num_partitions=3, replication_factor=1)])
    try:
        result.all().get()                  # one future per topic: result.values()
    except ExecutionError as e:
        if not isinstance(e.__cause__, TopicExistsError):
            raise
```

- Every call returns its `…Result` at once, as Java's does; the Result holds one
  `KafkaFuture` per key (or one for the whole call) and `.get()` waits. A failed key
  raises `ExecutionError`, whose `__cause__` is the error as its class.
- The sync client has one daemon pump thread that completes its futures, as Java's
  admin thread does: `then_apply` functions and `when_complete` actions run there,
  so a blocking action blocks the client's other completions.
- On the async peer the calls are plain `def` too, and the Result holds
  `AsyncKafkaFuture`s, whose `get()` is awaited; their actions run on the event loop:

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

  async with AsyncAdmin.create(configs={"bootstrap.servers": "localhost:9092"}) as admin:
      result = admin.create_topics(new_topics=[NewTopic(name="demo", num_partitions=3, replication_factor=1)])
      await result.all().get()
  ```

- Errors are classes: `isinstance(e.__cause__, RetriableError)`.
- `MockAdminClient` (and `AsyncMockAdminClient`) is Java's `MockAdminClient`, for tests: no
  broker, in-memory state, `MockAdminClient.create()...build()` (`MockAdminClient()` has one `Node.no_node()` broker, as Java's, so topic calls need real brokers). Where Java's mock
  throws "Not implemented yet", it raises `UnsupportedOperationError` (Java's
  `UnsupportedOperationException`, a `RuntimeError` at the package root, not a
  `KafkaError`).
- Overloads whose entry point the core lacks (streams and share groups, metadata quorum, …)
  are not generated; they are listed in the docstring of `confluent_kafka.admin.admin`.

## Errors

Errors are a typed hierarchy caught by class, not a flat error with a `code()`.
There is no `code()` / `is_retriable()` / `is_fatal()` on the surface —
retriability is `except RetriableError`, and each Java exception class has a
`…Error` Python analog in the module of its Java package (generated from the
Java sources). The Java built-in exception classes (`IllegalStateError`,
`IllegalArgumentError`, `ConcurrentModificationError`, `TimeoutError`,
`NoSuchElementError`, `ExecutionError`, `UnsupportedOperationError`,
`IndexOutOfBoundsError`) live at the package root. `UnsupportedOperationError` is Java's `UnsupportedOperationException`,
which the mocks raise for what Java's mocks do not implement
(`MockConsumer.offsets_for_times`, `MockAdminClient`'s "Not implemented yet"), and the
client's core for what this client does not support yet; it is not a `KafkaError`, so
`except KafkaError` does not catch it, as in Java.

## Directory Structure

```
confluent_kafka/        # The client package, one module per Java package
_confluentkafka.c       # Hand-written CPython extension over the C API
tinycthread.{c,h}, c11threads_compat.h  # C11 threads shim (macOS)
setup.py, pyproject.toml
grpc_server.py, grpc_server_async.py, grpc_translate.py
                        # gRPC servers for the multilanguage tests
Dockerfile.grpc, Dockerfile.grpc.async  # Their images
examples/               # Runnable producer, consumer and admin examples
test/
├── unit/               # pytest unit tests, no broker needed
├── integration/        # Broker-backed binding tests (testcontainers; skip without Docker)
├── static/             # Source checks: format arity
└── performance/        # Latency-budget and benchmark tests (need a broker)
tools/                  # format_arity.py
soak/                   # Long-running soak client — see soak/README.md
Makefile                # Targets the repository root's Makefile delegates to
```

## Where commands run

Run the commands below **from this directory**, inside a Python environment
(3.10+). The repository root's `make` targets use the shared `../venv/`,
which `make init` at the root creates.

## Prerequisites

The extension links the Rust library and includes its C header, so build the
Rust client first:

```bash
make -C ../rust build-all-features   # or: cd ../rust && cargo build --all-features --release
```

This produces `../rust/target/release/libconfluent_kafka.*` and
`../rust/target/include/confluent_kafka.h`. `setup.py` finds them there;
set `CONFLUENT_KAFKA_LIB_DIR` to link a library from another directory
(e.g. `../rust/target/debug`).

## Building

```bash
pip install -e .          # builds _confluentkafka in place
pip install .[dev]        # test and benchmark dependencies
```

## Running Tests

```bash
python -m pytest test/unit      # unit tests
python -m pytest test/integration   # broker-backed tests (Docker)
python -m pytest test/static    # static checks of the extension source
python -m pytest soak/test      # soak client unit tests
python -m mypy --strict confluent_kafka test/unit/test_typing.py   # typecheck
```

The static checks read source files only, so they need neither the Rust
library nor the extension built.

The Python arms of the multilanguage integration suite (`__grpc_python`,
`__grpc_python_async`) run the Rust test harness against the gRPC servers
above; `make test-integration` builds the images and runs them (Linux only,
Docker required).

## Make targets

`Makefile` wraps the commands above. The repository root's Makefile delegates
to it (activating the venv and building the Rust library first); it can also
be run directly:

```bash
make build                # pip install -e . against ../rust/target/$(PROFILE)
make test                 # typecheck, unit, integration, static and soak tests, then test-integration
make test-unit            # unit tests only
make typecheck            # mypy --strict over the package and test/unit/test_typing.py
make check-static         # static checks only
make test-integration     # gRPC images + the __grpc_python arms (Linux, Docker)
make test-performance     # latency-budget tests (idle machine, broker needed)
make consumer-perf-test producer-perf-test   # env-driven benchmarks
make grpc-image grpc-image-async
make clean
```

`PROFILE` (`release` by default, or `debug`) selects which Rust build to link.
`CLIENT_VERSION=2` runs the benchmarks against the PyPI librdkafka client in
its own venv (`make init-venv-librdkafka` at the root): it installs the same
top-level `confluent_kafka` package as this client, so the two cannot share one.

## License

Apache License 2.0. See [LICENSE](../LICENSE) for details.
