Ignore MetadataResponses with empty broker list (#1506)

Author: dpkp

kafka/cluster.py

@@ -214,7 +214,8 @@ def update_metadata(self, metadata):
             return self.failed_update(error)
 
         if not metadata.brokers:
-            log.warning("No broker metadata found in MetadataResponse")
+            log.warning("No broker metadata found in MetadataResponse -- ignoring.")
+            return self.failed_update(Errors.MetadataEmptyBrokerList(metadata))
 
         _new_brokers = {}
         for broker in metadata.brokers:

kafka/errors.py

@@ -54,6 +54,10 @@ class StaleMetadata(KafkaError):
     invalid_metadata = True
 
 
+class MetadataEmptyBrokerList(KafkaError):
+    retriable = True
+
+
 class UnrecognizedBrokerVersion(KafkaError):
     pass
 

test/test_cluster.py

@@ -0,0 +1,22 @@
+# pylint: skip-file
+from __future__ import absolute_import
+
+import pytest
+
+from kafka.cluster import ClusterMetadata
+from kafka.protocol.metadata import MetadataResponse
+
+
+def test_empty_broker_list():
+    cluster = ClusterMetadata()
+    assert len(cluster.brokers()) == 0
+
+    cluster.update_metadata(MetadataResponse[0](
+        [(0, 'foo', 12), (1, 'bar', 34)], []))
+    assert len(cluster.brokers()) == 2
+
+    # empty broker list response should be ignored
+    cluster.update_metadata(MetadataResponse[0](
+        [],  # empty brokers
+        [(17, 'foo', []), (17, 'bar', [])]))  # topics w/ error
+    assert len(cluster.brokers()) == 2