From c9bff0701ea153933dee8c03cf8feb3470e39c52 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sun, 22 May 2016 11:35:26 -0700 Subject: Use Fetch/Produce API v2 for brokers >= 0.10 (uses message format v1) --- kafka/producer/kafka.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'kafka/producer/kafka.py') diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 7aa24b3..fc60e78 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -283,7 +283,8 @@ class KafkaProducer(object): if self.config['compression_type'] == 'lz4': assert self.config['api_version'] >= (0, 8, 2), 'LZ4 Requires >= Kafka 0.8.2 Brokers' - self._accumulator = RecordAccumulator(**self.config) + message_version = 1 if self.config['api_version'] >= (0, 10) else 0 + self._accumulator = RecordAccumulator(message_version=message_version, **self.config) self._metadata = client.cluster self._sender = Sender(client, self._metadata, self._accumulator, **self.config) -- cgit v1.2.1