Minor refactor in conn.py, update version in __init__.py, add ErrorString

Author: Mark Roberts

kafka/client.py

@@ -5,7 +5,7 @@
 from functools import partial
 from itertools import count
 
-from kafka.common import (ErrorMapping, TopicAndPartition,
+from kafka.common import (ErrorMapping, ErrorStrings, TopicAndPartition,
                           ConnectionError, FailedPayloadsError,
                           BrokerResponseError, PartitionUnavailableError,
                           KafkaUnavailableError, KafkaRequestError)
@@ -174,8 +174,8 @@ def _raise_on_response_error(self, resp):
             self.reset_topic_metadata(resp.topic)
 
         raise BrokerResponseError(
-            "Request for %s failed with errorcode=%d" %
-            (TopicAndPartition(resp.topic, resp.partition), resp.error))
+            "Request for %s failed with errorcode=%d (%s)" %
+            (TopicAndPartition(resp.topic, resp.partition), resp.error, ErrorStrings[resp.error]))
 
     #################
     #   Public API  #

kafka/common.py

@@ -48,22 +48,28 @@
 TopicAndPartition = namedtuple("TopicAndPartition", ["topic", "partition"])
 
 
+ErrorStrings = {
+    -1 : 'UNKNOWN',
+    0 : 'NO_ERROR',
+    1 : 'OFFSET_OUT_OF_RANGE',
+    2 : 'INVALID_MESSAGE',
+    3 : 'UNKNOWN_TOPIC_OR_PARTITON',
+    4 : 'INVALID_FETCH_SIZE',
+    5 : 'LEADER_NOT_AVAILABLE',
+    6 : 'NOT_LEADER_FOR_PARTITION',
+    7 : 'REQUEST_TIMED_OUT',
+    8 : 'BROKER_NOT_AVAILABLE',
+    9 : 'REPLICA_NOT_AVAILABLE',
+    10 : 'MESSAGE_SIZE_TO_LARGE',
+    11 : 'STALE_CONTROLLER_EPOCH',
+    12 : 'OFFSET_METADATA_TOO_LARGE',
+}
+
 class ErrorMapping(object):
-    # Many of these are not actually used by the client
-    UNKNOWN                   = -1
-    NO_ERROR                  = 0
-    OFFSET_OUT_OF_RANGE       = 1
-    INVALID_MESSAGE           = 2
-    UNKNOWN_TOPIC_OR_PARTITON = 3
-    INVALID_FETCH_SIZE        = 4
-    LEADER_NOT_AVAILABLE      = 5
-    NOT_LEADER_FOR_PARTITION  = 6
-    REQUEST_TIMED_OUT         = 7
-    BROKER_NOT_AVAILABLE      = 8
-    REPLICA_NOT_AVAILABLE     = 9
-    MESSAGE_SIZE_TO_LARGE     = 10
-    STALE_CONTROLLER_EPOCH    = 11
-    OFFSET_METADATA_TOO_LARGE = 12
+    pass
+
+for k, v in ErrorStrings.items():
+    setattr(ErrorMapping, v, k)
 
 #################
 #   Exceptions  #

kafka/conn.py

@@ -28,11 +28,10 @@ def __init__(self, host, port, timeout=DEFAULT_SOCKET_TIMEOUT_SECONDS):
         super(KafkaConnection, self).__init__()
         self.host = host
         self.port = port
-        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self._sock.connect((host, port))
         self.timeout = timeout
-        self._sock.settimeout(self.timeout)
-        self._dirty = False
+        self._sock = None
+
+        self.reinit()
 
     def __repr__(self):
         return "<KafkaConnection host=%s port=%d>" % (self.host, self.port)
@@ -47,24 +46,28 @@ def _raise_connection_error(self):
 
     def _read_bytes(self, num_bytes):
         bytes_left = num_bytes
-        resp = ''
+        responses = []
+
         log.debug("About to read %d bytes from Kafka", num_bytes)
         if self._dirty:
             self.reinit()
+
         while bytes_left:
             try:
-                data = self._sock.recv(bytes_left)
+                data = self._sock.recv(min(bytes_left, 4096))
             except socket.error:
                 log.exception('Unable to receive data from Kafka')
                 self._raise_connection_error()
+
             if data == '':
                 log.error("Not enough data to read this response")
                 self._raise_connection_error()
+
             bytes_left -= len(data)
             log.debug("Read %d/%d bytes from Kafka", num_bytes - bytes_left, num_bytes)
-            resp += data
+            responses.append(data)
 
-        return resp
+        return ''.join(responses)
 
     ##################
     #   Public API   #

load_example.py

@@ -53,5 +53,5 @@ def main():
     print 'Messages invalid: %d' % threads[1].invalid
 
 if __name__ == "__main__":
-    logging.basicConfig(level=logging.DEBUG)
+    logging.basicConfig(level=logging.WARN)
     main()