Backoff on unavailable group coordinator retry (dpkp#1125)

Author: dpkp

kafka/coordinator/base.py

@@ -219,6 +219,8 @@ def ensure_coordinator_known(self):
                         log.debug('Requesting metadata for group coordinator request: %s', future.exception)
                         metadata_update = self._client.cluster.request_update()
                         self._client.poll(future=metadata_update)
+                    else:
+                        time.sleep(self.config['retry_backoff_ms'] / 1000)
                 else:
                     raise future.exception  # pylint: disable-msg=raising-bad-type