Update kafka-src to latest 0.8

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

Author: mumrah

kafka-src

@@ -116,7 +116,7 @@ def _send_broker_aware_request(self, payloads, encoder_fn, decoder_fn):
         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 = {}
 

kafka/client.py

@@ -71,8 +71,9 @@ def _consume_response_iter(self):
 
     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()
 

kafka/conn.py

@@ -225,14 +225,25 @@ def test_produce_mixed(self):
 
 
     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)