confluentinc / confluentinc/confluent-kafka-python

produce_batch(): support headers and timestamps (headers are currently dropped silently)

Open
#2,320 0 comments 0 reactions 1 assignee Claimed by @ojasvajain View on GitHub
status:waiting-for-interest
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.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.