Update kafka-src to latest 0.8

Fix a broken test (100k was too much to send in one batch)

4 files changed, 17 insertions, 5 deletions

diff --git a/kafka-src b/kafka-src
-Subproject 30f992c52dd3aba8cc35c9835df5b5e56581e0d
+Subproject 3c27988ca4036985f4c7bef62b9bbae3f95f0fb
diff --git a/kafka/client.py b/kafka/client.py
index 23b0a48..1c7fc93 100644
--- a/kafka/client.py
+++ b/kafka/client.py
@@ -116,7 +116,7 @@ class KafkaClient(object):
         for payload in payloads:
             payloads_by_broker[self._get_leader_for_partition(payload.topic, payload.partition)].append(payload)
             original_keys.append((payload.topic, payload.partition))
-        
+
         # Accumulate the responses in a dictionary
         acc = {}
 
diff --git a/kafka/conn.py b/kafka/conn.py
index 203488b..c9ba4cc 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -71,8 +71,9 @@ class KafkaConnection(object):
 
     def send(self, requestId, payload):
         "Send a request to Kafka"
+        log.debug("About to send %d bytes to Kafka" % len(payload))
         sent = self._sock.sendall(payload)
-        if sent == 0:
+        if sent != None:
             raise RuntimeError("Kafka went away")
         self.data = self._consume_response()
 
diff --git a/test/integration.py b/test/integration.py
index 04d5979..a1fcce7 100644
--- a/test/integration.py
+++ b/test/integration.py
@@ -225,15 +225,26 @@ class TestKafkaClient(unittest.TestCase):
 
 
     def test_produce_100k_gzipped(self):
-        produce = ProduceRequest("test_produce_100k_gzipped", 0, messages=[
-            create_gzip_message(["Gzipped %d" % i for i in range(100000)])
+        req1 = ProduceRequest("test_produce_100k_gzipped", 0, messages=[
+            create_gzip_message(["Gzipped batch 1, message %d" % i for i in range(50000)])
         ])
 
-        for resp in self.client.send_produce_request([produce]):
+        for resp in self.client.send_produce_request([req1]):
             self.assertEquals(resp.error, 0)
             self.assertEquals(resp.offset, 0)
 
         (offset, ) = self.client.send_offset_request([OffsetRequest("test_produce_100k_gzipped", 0, -1, 1)])
+        self.assertEquals(offset.offsets[0], 50000)
+
+        req2 = ProduceRequest("test_produce_100k_gzipped", 0, messages=[
+            create_gzip_message(["Gzipped batch 2, message %d" % i for i in range(50000)])
+        ])
+
+        for resp in self.client.send_produce_request([req2]):
+            self.assertEquals(resp.error, 0)
+            self.assertEquals(resp.offset, 50000)
+
+        (offset, ) = self.client.send_offset_request([OffsetRequest("test_produce_100k_gzipped", 0, -1, 1)])
         self.assertEquals(offset.offsets[0], 100000)
 
     #####################