Merge branch 'master' into conn_refactor

Author: Mark Roberts

kafka/client.py

@@ -83,14 +83,14 @@ def _send_broker_unaware_request(self, requestId, request):
         brokers. Keep trying until you succeed.
         """
         for (host, port) in self.hosts:
-            conn = self._get_conn(host, port)
             try:
+                conn = self._get_conn(host, port)
                 conn.send(requestId, request)
                 response = conn.recv(requestId)
                 return response
             except Exception, e:
-                log.warning("Could not send request [%r] to server %s, "
-                            "trying next server: %s" % (request, conn, e))
+                log.warning("Could not send request [%r] to server %s:%i, "
+                            "trying next server: %s" % (request, host, port, e))
                 continue
 
         raise KafkaUnavailableError("All servers failed to process request")

kafka/conn.py

@@ -15,11 +15,11 @@
 
 def collect_hosts(hosts, randomize=True):
     """
-    Collects a comma-separated set of hosts (host:port) and optionnaly
+    Collects a comma-separated set of hosts (host:port) and optionally
     randomize the returned list.
     """
 
-    if isinstance(hosts, str):
+    if isinstance(hosts, basestring):
         hosts = hosts.strip().split(',')
 
     result = []

test/test_unit.py

@@ -449,6 +449,16 @@ def test_init_with_csv(self):
             [('kafka01', 9092), ('kafka02', 9092), ('kafka03', 9092)],
             client.hosts)
 
+    def test_init_with_unicode_csv(self):
+
+        with patch.object(KafkaClient, 'load_metadata_for_topics'):
+            client = KafkaClient(
+                hosts=u'kafka01:9092,kafka02:9092,kafka03:9092')
+
+        self.assertItemsEqual(
+            [('kafka01', 9092), ('kafka02', 9092), ('kafka03', 9092)],
+            client.hosts)
+
     def test_send_broker_unaware_request_fail(self):
         'Tests that call fails when all hosts are unavailable'