Merge pull request #611 from dpkp/sock_send_bytes

Handle partial socket send()

2 files changed, 8 insertions, 6 deletions

diff --git a/kafka/conn.py b/kafka/conn.py
index 2b82b6d..ffc839e 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -188,10 +188,12 @@ class BrokerConnection(object):
             # and send bytes asynchronously. For now, just block
             # sending each request payload
             self._sock.setblocking(True)
-            sent_bytes = self._sock.send(size)
-            assert sent_bytes == len(size)
-            sent_bytes = self._sock.send(message)
-            assert sent_bytes == len(message)
+            for data in (size, message):
+                total_sent = 0
+                while total_sent < len(data):
+                    sent_bytes = self._sock.send(data[total_sent:])
+                    total_sent += sent_bytes
+                assert total_sent == len(data)
             self._sock.setblocking(False)
         except (AssertionError, ConnectionError) as e:
             log.exception("Error sending %s to %s", request, self)
diff --git a/kafka/future.py b/kafka/future.py
index 06b8c3a..c7e0b14 100644
--- a/kafka/future.py
+++ b/kafka/future.py
@@ -15,10 +15,10 @@ class Future(object):
         self._errbacks = []
 
     def succeeded(self):
-        return self.is_done and not self.exception
+        return self.is_done and not bool(self.exception)
 
     def failed(self):
-        return self.is_done and self.exception
+        return self.is_done and bool(self.exception)
 
     def retriable(self):
         try: