Merge pull request #811 from Parsely/bugfix/group_coordinator_discove…

…ry_logging improve logging and retry logic when broker is unreachable
Parsely · May 29, 2018 · 56efe39 · 56efe39
2 parents 8585540 + 899885a
commit 56efe39
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 4 deletions.
diff --git a/pykafka/broker.py b/pykafka/broker.py
@@ -120,8 +120,9 @@ def __init__(self,
         try:
             self.connect()
         except SocketDisconnectedError:
-            log.warning("Failed to connect newly created broker for {host}:{port}".format(
-                host=self._host, port=self._port))
+            log.warning("Failed to connect to broker at {host}:{port}. Check the "
+                        "`listeners` property in server.config."
+                        .format(host=self._host, port=self._port))
 
     def __repr__(self):
         return "<{module}.{name} at {id_} (host={host}, port={port}, id={my_id})>".format(

diff --git a/pykafka/cluster.py b/pykafka/cluster.py
@@ -453,11 +453,18 @@ def get_group_coordinator(self, consumer_group):
             for broker in itervalues(self.brokers):
 
                 req = GroupCoordinatorRequest(consumer_group)
-                future = broker.handler.request(req)
+                try:
+                    future = broker.handler.request(req)
+                except AttributeError:
+                    log.error("Broker {} not connected during offset manager discovery"
+                              .format(broker.id))
+                    if i == max_connection_retries - 1:
+                        raise
+                    continue
                 try:
                     res = future.get(GroupCoordinatorResponse)
                 except GroupCoordinatorNotAvailable:
-                    log.error('Error discovering offset manager.')
+                    log.error('Error discovering offset manager - GroupCoordinatorNotAvailable.')
                     if i == max_connection_retries - 1:
                         raise
                 except SocketDisconnectedError: