Fix KafkaConsumer autocommit for 0.8 brokers (#756 / #706)

  * Dont wait for group join to enable AutoCommitTask if broker version < 0.9
  * For zookeeper offset storage, set a "coordinator" with least_loaded_node

1 files changed, 9 insertions, 0 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)