confluentinc / confluentinc/confluent-kafka-python

Producer.poll(timeout) blocks for full timeout after delivery callback fires (regression in 2.13.0)

Open
#2,215 1 comment 3 reactions 0 assignees View on GitHub
component:producer
Dominant language
Python
Stars
509
Forks
964
Avg merge
2d 2h
Merged PRs (30d)
14

Description

# Description

`Producer.poll(timeout)` no longer returns early after processing delivery callbacks when timeout > 200ms. Instead it blocks for the entire timeout duration, regardless of whether events were served.

This is a regression introduced in 2.13.0 by the "wakeable poll" refactor (PR #2126 ). The original single-call implementation of `rd_kafka_poll()` returned as soon as events were processed. The new chunked loop replicates this correctly for `Consumer.poll()` but not for `Producer.poll()`.

The producer loop accumulates `chunk_result` but never breaks when `chunk_result > 0`, causing it to exhaust all remaining chunks even after the delivery callback has fired.

# Step to reproduce
```python
import time
from confluent_kafka import Producer, __version__

KAFKA_BOOTSTRAP = "localhost:9092"
POLL_TIMEOUT = 5.0

print(f"confluent-kafka version: {__version__}")

producer = Producer({"bootstrap.servers": KAFKA_BOOTSTRAP})

def on_delivery(err, msg):
print(f" [{time.time():.3f}] on_delivery fired — err={err}, topic={msg.topic()}")

print(f"[{time.time():.3f}] produce()")
producer.produce("test-topic", value=b"hello", on_delivery=on_delivery)

print(f"[{time.time():.3f}] poll({POLL_TIMEOUT}) start")
t0 = time.time()
producer.poll(POLL_TIMEOUT)
elapsed = time.time() - t0
print(f"[{time.time():.3f}] poll() returned after {elapsed:.3f}s")
```

## Output in 2.12.2

```python
confluent-kafka version: 2.12.2
[1733305268.206] produce()
[1733305268.206] poll(5.0) start
[1733305268.532] on_delivery fired — err=None, topic=test-topic
[1733305268.532] poll() returned after 0.325s
```

## Output in 2.13.2

```python
confluent-kafka version: 2.13.2
[1733305281.264] produce()
[1733305281.265] poll(5.0) start
[1733305281.276] on_delivery fired — err=None, topic=test-topic
[1733305286.078] poll() returned after 4.813s
```

# Potential Cause

In `Producer_poll0` (`src/confluent_kafka/src/Producer.c`), the wakeable loop does not break when `chunk_result > 0`

# Potential fix

Add an early exit in `Producer_poll0` after events are served, matching the original `rd_kafka_poll()` single-call behaviour and the consumer implementation:

```python
int chunk_result = rd_kafka_poll(self->rk, chunk_timeout_ms);
if (chunk_result < 0) { r = chunk_result; break; }

r += chunk_result;
if (chunk_result > 0) break; // ← return early once events are served
```

The new chunked implementation was intended to preserve this early-exit behaviour while adding signal interruptibility. It does so correctly for `Consumer.poll()` (`if (rkm) break;`) but the equivalent guard was omitted in `Producer_poll0`, making `poll(5.0)` in 2.13.x strictly worse than `poll(5.0)` in 2.12.x when the broker is fast.

# Environment

- confluent-kafka-python: 2.13.0, 2.13.1, 2.13.2
- Python: 3.11+
- OS: Linux

---

Thanks to the maintainers for the great work on confluent kafka. Happy to open a PR with the proposed fix if the issue is
confirmed.

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.