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:
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.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.
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 modeChecklist
Description
Request: support
headersandtimestampinProducer.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:headersare silently discarded — no exception, no_error, the message counts as queued.correlation_id,content-typeand tracing context vanish, and consumer-side parsing breaks at runtime with no signal.timestampraisesNotImplementedError, although the docstring lists'timestamp' (int)among accepted keys — docs and implementation disagree.Upstream cause: (1) is inherited from librdkafka, where
rd_kafka_message_thas 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.Checklist
2.14.0/('2.14.0', 34472191){'bootstrap.servers': 'localhost:1', 'queue.buffering.max.messages': 100}