Handle lookup_coordinator send failures (#1279)

Author: dpkp

kafka/coordinator/base.py

@@ -228,11 +228,17 @@ def _reset_find_coordinator_future(self, result):
         self._find_coordinator_future = None
 
     def lookup_coordinator(self):
-        if self._find_coordinator_future is None:
-            self._find_coordinator_future = self._send_group_coordinator_request()
-
-            self._find_coordinator_future.add_both(self._reset_find_coordinator_future)
-        return self._find_coordinator_future
+        if self._find_coordinator_future is not None:
+            return self._find_coordinator_future
+
+        # If there is an error sending the group coordinator request
+        # then _reset_find_coordinator_future will immediately fire and
+        # set _find_coordinator_future = None
+        # To avoid returning None, we capture the future in a local variable
+        self._find_coordinator_future = self._send_group_coordinator_request()
+        future = self._find_coordinator_future
+        self._find_coordinator_future.add_both(self._reset_find_coordinator_future)
+        return future
 
     def need_rejoin(self):
         """Check whether the group should be rejoined (e.g. if metadata changes)

test/test_coordinator.py

@@ -590,3 +590,11 @@ def test_heartbeat(patched_coord):
     patched_coord.heartbeat_task()
     assert patched_coord._client.schedule.call_count == 1
     assert patched_coord.heartbeat_task._handle_heartbeat_failure.call_count == 1
+
+
+def test_lookup_coordinator_failure(mocker, coordinator):
+
+    mocker.patch.object(coordinator, '_send_group_coordinator_request',
+                        return_value=Future().failure(Exception('foobar')))
+    future = coordinator.lookup_coordinator()
+    assert future.failed()