From 08c77499a2e8bc79d6788d70ef96d77752ed6325 Mon Sep 17 00:00:00 2001 From: Heikki Nousiainen Date: Tue, 14 Aug 2018 15:17:23 +0300 Subject: Support produce with Kafka record headers --- test/test_producer.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) (limited to 'test') diff --git a/test/test_producer.py b/test/test_producer.py index 09d184f..176b239 100644 --- a/test/test_producer.py +++ b/test/test_producer.py @@ -91,10 +91,16 @@ def test_kafka_producer_proper_record_metadata(kafka_broker, compression): compression_type=compression) magic = producer._max_usable_produce_magic() + # record headers are supported in 0.11.0 + if version() < (0, 11, 0): + headers = None + else: + headers = [("Header Key", b"Header Value")] + topic = random_string(5) future = producer.send( topic, - value=b"Simple value", key=b"Simple key", timestamp_ms=9999999, + value=b"Simple value", key=b"Simple key", headers=headers, timestamp_ms=9999999, partition=0) record = future.get(timeout=5) assert record is not None @@ -116,6 +122,8 @@ def test_kafka_producer_proper_record_metadata(kafka_broker, compression): assert record.serialized_key_size == 10 assert record.serialized_value_size == 12 + if headers: + assert record.serialized_header_size == 22 # generated timestamp case is skipped for broker 0.9 and below if magic == 0: -- cgit v1.2.1