Fix more tests, only multiprocessing consumer ones remaining

2 files changed, 7 insertions, 3 deletions

diff --git a/kafka/client.py b/kafka/client.py
index e098470..410573a 100644
--- a/kafka/client.py
+++ b/kafka/client.py
@@ -145,7 +145,7 @@ class KafkaClient(object):
 
         # For each broker, send the list of request payloads
         for broker, payloads in payloads_by_broker.items():
-            conn = self._get_conn(broker.host, broker.port)
+            conn = self._get_conn(broker.host.decode('utf-8'), broker.port)
             requestId = self._next_id()
             request = encoder_fn(client_id=self.client_id,
                                  correlation_id=requestId, payloads=payloads)
@@ -233,8 +233,8 @@ class KafkaClient(object):
         A reinit() has to be done on the copy before it can be used again
         """
         c = copy.deepcopy(self)
-        for k, v in c.conns.items():
-            c.conns[k] = v.copy()
+        for key in c.conns:
+            c.conns[key] = self.conns[key].copy()
         return c
 
     def reinit(self):
diff --git a/kafka/conn.py b/kafka/conn.py
index 41cd424..aef0299 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -155,6 +155,10 @@ class KafkaConnection(local):
         return a new KafkaConnection object
         """
         c = copy.deepcopy(self)
+        # Python 3 doesn't copy custom attributes of the threadlocal subclass
+        c.host = copy.copy(self.host)
+        c.port = copy.copy(self.port)
+        c.timeout = copy.copy(self.timeout)
         c._sock = None
         return c