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

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)
 

kafka/coordinator/consumer.py

@@ -100,6 +100,12 @@ class ConsumerCoordinator(BaseCoordinator):
                 interval = self.config['auto_commit_interval_ms'] / 1000.0
                 self._auto_commit_task = AutoCommitTask(weakref.proxy(self), interval)
 
+                # When using broker-coordinated consumer groups, auto-commit will
+                # be automatically enabled on group join (see _on_join_complete)
+                # Otherwise, we should enable now b/c there will be no group join
+                if self.config['api_version'] < (0, 9):
+                    self._auto_commit_task.enable()
+
         self._sensors = ConsumerCoordinatorMetrics(metrics, metric_group_prefix,
                                                    self._subscription)
 
@@ -293,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)
@@ -356,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)
@@ -415,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)
@@ -571,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):

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)