Changes based on comments by @rdiomar, plus added LeaderUnavailableError for clarity

Author: mrtheb

kafka/client.py

@@ -8,6 +8,7 @@
 from kafka.common import (ErrorMapping, TopicAndPartition,
                           ConnectionError, FailedPayloadsError,
                           BrokerResponseError, PartitionUnavailableError,
+                          LeaderUnavailableError,
                           KafkaUnavailableError)
 
 from kafka.conn import KafkaConnection, DEFAULT_SOCKET_TIMEOUT_SECONDS
@@ -124,8 +125,10 @@ def _send_broker_aware_request(self, payloads, encoder_fn, decoder_fn):
             leader = self._get_leader_for_partition(payload.topic,
                                                     payload.partition)
             if leader is None:
-                raise PartitionUnavailableError(
-                    "No leader for topic %s partition %s" % (payload.topic, payload.partition))
+                raise LeaderUnavailableError(
+                    "Leader not available for topic %s partition %s" %
+                    (payload.topic, payload.partition))
+
             payloads_by_broker[leader].append(payload)
             original_keys.append((payload.topic, payload.partition))
 
@@ -250,15 +253,15 @@ def load_metadata_for_topics(self, *topics):
             self.reset_topic_metadata(topic)
 
             if not partitions:
-                log.info('No partitions for %s', topic)
+                log.warning('No partitions for %s', topic)
                 continue
 
             self.topic_partitions[topic] = []
             for partition, meta in partitions.items():
                 self.topic_partitions[topic].append(partition)
                 topic_part = TopicAndPartition(topic, partition)
                 if meta.leader == -1:
-                    log.info('No leader for topic %s partition %s', topic, partition)
+                    log.warning('No leader for topic %s partition %s', topic, partition)
                     self.topics_to_brokers[topic_part] = None
                 else:
                     self.topics_to_brokers[topic_part] = brokers[meta.leader]

kafka/common.py

@@ -82,6 +82,10 @@ class BrokerResponseError(KafkaError):
     pass
 
 
+class LeaderUnavailableError(KafkaError):
+    pass
+
+
 class PartitionUnavailableError(KafkaError):
     pass
 

test/test_unit.py

@@ -7,7 +7,7 @@
     ProduceRequest, FetchRequest, Message, ChecksumError,
     ConsumerFetchSizeTooSmall, ProduceResponse, FetchResponse,
     OffsetAndMessage, BrokerMetadata, PartitionMetadata,
-    TopicAndPartition, PartitionUnavailableError
+    TopicAndPartition, LeaderUnavailableError, PartitionUnavailableError
 )
 from kafka.codec import (
     has_gzip, has_snappy, gzip_encode, gzip_decode,
@@ -386,7 +386,7 @@ def test_decode_offset_fetch_response(self):
         pass
 
 
-class TestClient(unittest.TestCase):
+class TestKafkaClient(unittest.TestCase):
 
     @patch('kafka.client.KafkaConnection')
     @patch('kafka.client.KafkaProtocol')
@@ -520,7 +520,7 @@ def test_get_leader_returns_none_when_noleader(self, protocol, conn):
     @patch('kafka.client.KafkaConnection')
     @patch('kafka.client.KafkaProtocol')
     def test_send_produce_request_raises_when_noleader(self, protocol, conn):
-        "Getting leader for partitions returns None when the partiion has no leader"
+        "Send producer request raises LeaderUnavailableError if leader is not available"
 
         conn.recv.return_value = 'response'  # anything but None
 
@@ -542,7 +542,7 @@ def test_send_produce_request_raises_when_noleader(self, protocol, conn):
             [create_message("a"), create_message("b")])]
 
         self.assertRaises(
-            PartitionUnavailableError,
+            LeaderUnavailableError,
             client.send_produce_request, requests)
 
 if __name__ == '__main__':