diff options
| author | Dana Powers <dana.powers@gmail.com> | 2016-05-22 12:41:17 -0700 |
|---|---|---|
| committer | Dana Powers <dana.powers@gmail.com> | 2016-05-22 12:41:17 -0700 |
| commit | 96530f6a9c4a31d23b069ba162dba6cf45a5efd0 (patch) | |
| tree | e9cc1eddfd4f03981762127d035c58db9c9a5269 /kafka/producer | |
| parent | 7941a2ac7ec6663f08c6291d92746eae9f792916 (diff) | |
| download | kafka-python-96530f6a9c4a31d23b069ba162dba6cf45a5efd0.tar.gz | |
Use Fetch/Produce API v2 for brokers >= 0.10 (uses message format v1) (#694)
Diffstat (limited to 'kafka/producer')
| -rw-r--r-- | kafka/producer/kafka.py | 3 | ||||
| -rw-r--r-- | kafka/producer/sender.py | 16 |
2 files changed, 15 insertions, 4 deletions
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) diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py index 9c36c9b..f10c34c 100644 --- a/kafka/producer/sender.py +++ b/kafka/producer/sender.py @@ -174,11 +174,16 @@ class Sender(threading.Thread): for batch in batches]) for topic, partitions in response.topics: - for partition, error_code, offset in partitions: + for partition_info in partitions: + if response.API_VERSION < 2: + partition, error_code, offset = partition_info + ts = None + else: + partition, error_code, offset, ts = partition_info tp = TopicPartition(topic, partition) error = Errors.for_code(error_code) batch = batches_by_partition[tp] - self._complete_batch(batch, error, offset) + self._complete_batch(batch, error, offset, ts) else: # this is the acks = 0 case, just complete all requests @@ -258,7 +263,12 @@ class Sender(threading.Thread): buf = batch.records.buffer() produce_records_by_partition[topic][partition] = buf - version = 1 if self.config['api_version'] >= (0, 9) else 0 + if self.config['api_version'] >= (0, 10): + version = 2 + elif self.config['api_version'] == (0, 9): + version = 1 + else: + version = 0 return ProduceRequest[version]( required_acks=acks, timeout=timeout, |