confluentinc / confluentinc/confluent-kafka-python
produce_batch(): support headers and timestamps (headers are currently dropped silently)
- Dominant language
- Python
- Stars
- 509
- Forks
- 964
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 14
Description
Description
===========
**Request:** support `headers` and `timestamp` in `Producer.produce_batch()`; until headers land, raise on them instead of dropping them.
**Why:** `produce_batch()` (added in v2.12.0, #2047) is the only API that enqueues N messages in one call and reports partial failure via `_error` + the queued count instead of raising all-or-nothing. Two gaps keep client libraries out:
1. **`headers` are silently discarded** — no exception, no `_error`, the message counts as queued. `correlation_id`, `content-type` and tracing context vanish, and consumer-side parsing breaks at runtime with no signal.
2. **`timestamp` raises `NotImplementedError`**, although the docstring lists `'timestamp' (int)` among accepted keys — docs and implementation disagree.
**Upstream cause:** (1) is inherited from librdkafka, where `rd_kafka_message_t` has no headers field on produce — filed as confluentinc/librdkafka#5559. This issue tracks the Python side. (2) is fixable here.
**Concrete impact:** blocks FastStream (ag2ai/faststream#2993) from using `produce_batch()` to fix a batch-cancellation bug (ag2ai/faststream#2836) — we always attach headers, so adopting it today would silently corrupt every published message.
How to reproduce
================
No broker needed — `produce()` only enqueues locally.
```python
from confluent_kafka import Producer
p = Producer({"bootstrap.servers": "localhost:1", "queue.buffering.max.messages": 100})
msgs = [{"value": b"a", "headers": [("k", b"v")]}]
print(p.produce_batch("t", msgs), msgs)
# 1 [{'value': b'a', 'headers': [('k', b'v')]}] -> queued=1, no _error, header never sent
p.produce_batch("t", [{"value": b"a", "timestamp": 1700000000000}])
# NotImplementedError: Message timestamps are not currently supported in batch mode
```
Checklist
=========
- [x] confluent-kafka-python and librdkafka version: `2.14.0` / `('2.14.0', 34472191)`
- [x] Apache Kafka broker version: N/A — client-side API limitation, broker independent
- [x] Client configuration: `{'bootstrap.servers': 'localhost:1', 'queue.buffering.max.messages': 100}`
- [x] Operating system: macOS 15 (arm64), Python 3.11
- [x] Provide client logs: N/A — headers produce no error at all; that is the defect
- [x] Provide broker log excerpts: N/A — nothing reaches the broker
- [ ] Critical issue
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.