diff options
| author | Dana Powers <dana.powers@gmail.com> | 2016-05-22 11:35:26 -0700 |
|---|---|---|
| committer | Dana Powers <dana.powers@gmail.com> | 2016-05-22 11:35:26 -0700 |
| commit | c9bff0701ea153933dee8c03cf8feb3470e39c52 (patch) | |
| tree | e9cc1eddfd4f03981762127d035c58db9c9a5269 /kafka/consumer/fetcher.py | |
| parent | 7941a2ac7ec6663f08c6291d92746eae9f792916 (diff) | |
| download | kafka-python-produce_fetch_v2.tar.gz | |
Use Fetch/Produce API v2 for brokers >= 0.10 (uses message format v1)produce_fetch_v2
Diffstat (limited to 'kafka/consumer/fetcher.py')
| -rw-r--r-- | kafka/consumer/fetcher.py | 7 |
1 files changed, 6 insertions, 1 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index bf59775..e5a165e 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -581,7 +581,12 @@ class Fetcher(six.Iterator): log.debug("Adding fetch request for partition %s at offset %d", partition, position) - 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 requests = {} for node_id, partition_data in six.iteritems(fetchable): requests[node_id] = FetchRequest[version]( |