For zookeeper offset storage, set a "coordinator" with least_loaded_nodezk_autocommit

3 files changed, 21 insertions, 23 deletions

diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index 168115a..25dd000 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -50,6 +50,7 @@ class BaseCoordinator(object):
         'session_timeout_ms': 30000,
         'heartbeat_interval_ms': 3000,
         'retry_backoff_ms': 100,
+        'api_version': (0, 9),
     }
 
     def __init__(self, client, **configs):
@@ -194,6 +195,14 @@ class BaseCoordinator(object):
         """
         while self.coordinator_unknown():
 
+            # Prior to 0.8.2 there was no group coordinator
+            # so we will just pick a node at random and treat
+            # it as the "coordinator"
+            if self.config['api_version'] < (0, 8, 2):
+                self.coordinator_id = self._client.least_loaded_node()
+                self._client.ready(self.coordinator_id)
+                continue
+
             future = self._send_group_coordinator_request()
             self._client.poll(future=future)
 
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):
diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index 15b915d..735d278 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -425,8 +425,7 @@ def test_send_offset_commit_request_fail(patched_coord, offsets):
     ((0, 9), OffsetCommitRequest[2])])
 def test_send_offset_commit_request_versions(patched_coord, offsets,
                                              api_version, req_type):
-    # assuming fixture sets coordinator=0, least_loaded_node=1
-    expect_node = 0 if api_version >= (0, 8, 2) else 1
+    expect_node = 0
     patched_coord.config['api_version'] = api_version
 
     patched_coord._send_offset_commit_request(offsets)
@@ -522,7 +521,7 @@ def test_send_offset_fetch_request_fail(patched_coord, partitions):
 def test_send_offset_fetch_request_versions(patched_coord, partitions,
                                             api_version, req_type):
     # assuming fixture sets coordinator=0, least_loaded_node=1
-    expect_node = 0 if api_version >= (0, 8, 2) else 1
+    expect_node = 0
     patched_coord.config['api_version'] = api_version
 
     patched_coord._send_offset_fetch_request(partitions)