From 7fc032ccd058eb16c77c2168ceea4052e2b3c264 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Tue, 12 Jul 2016 08:40:14 -0700 Subject: For zookeeper offset storage, set a "coordinator" with least_loaded_node --- kafka/coordinator/consumer.py | 30 ++++++++++-------------------- 1 file changed, 10 insertions(+), 20 deletions(-) (limited to 'kafka/coordinator/consumer.py') diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py index 44d4c6c..083a36a 100644 --- a/kafka/coordinator/consumer.py +++ b/kafka/coordinator/consumer.py @@ -299,8 +299,7 @@ class ConsumerCoordinator(BaseCoordinator): return {} while True: - if self.config['api_version'] >= (0, 8, 2): - self.ensure_coordinator_known() + self.ensure_coordinator_known() # contact coordinator to fetch committed offsets future = self._send_offset_fetch_request(partitions) @@ -362,8 +361,7 @@ class ConsumerCoordinator(BaseCoordinator): return while True: - if self.config['api_version'] >= (0, 8, 2): - self.ensure_coordinator_known() + self.ensure_coordinator_known() future = self._send_offset_commit_request(offsets) self._client.poll(future=future) @@ -421,14 +419,10 @@ class ConsumerCoordinator(BaseCoordinator): log.debug('No offsets to commit') return Future().success(True) - if self.config['api_version'] >= (0, 8, 2): - if self.coordinator_unknown(): - return Future().failure(Errors.GroupCoordinatorNotAvailableError) - node_id = self.coordinator_id - else: - node_id = self._client.least_loaded_node() - if node_id is None: - return Future().failure(Errors.NoBrokersAvailable) + elif self.coordinator_unknown(): + return Future().failure(Errors.GroupCoordinatorNotAvailableError) + + node_id = self.coordinator_id # create the offset commit request offset_data = collections.defaultdict(dict) @@ -577,14 +571,10 @@ class ConsumerCoordinator(BaseCoordinator): if not partitions: return Future().success({}) - if self.config['api_version'] >= (0, 8, 2): - if self.coordinator_unknown(): - return Future().failure(Errors.GroupCoordinatorNotAvailableError) - node_id = self.coordinator_id - else: - node_id = self._client.least_loaded_node() - if node_id is None: - return Future().failure(Errors.NoBrokersAvailable) + elif self.coordinator_unknown(): + return Future().failure(Errors.GroupCoordinatorNotAvailableError) + + node_id = self.coordinator_id # Verify node is ready if not self._client.ready(node_id): -- cgit v1.2.1