Bring acks and timeout down to the client

2 files changed, 5 insertions, 5 deletions

diff --git a/kafka/client.py b/kafka/client.py
index 862a30e..23b0a48 100644
--- a/kafka/client.py
+++ b/kafka/client.py
@@ -143,7 +143,7 @@ class KafkaClient(object):
         for conn in self.conns.values():
             conn.close()
 
-    def send_produce_request(self, payloads=[], fail_on_error=True, callback=None):
+    def send_produce_request(self, payloads=[], acks=1, timeout=1000, fail_on_error=True, callback=None):
         """
         Encode and send some ProduceRequests
 
@@ -162,8 +162,8 @@ class KafkaClient(object):
         list of ProduceResponse or callback(ProduceResponse), in the order of input payloads
         """
         resps = self._send_broker_aware_request(payloads,
-                                   KafkaProtocol.encode_produce_request,
-                                   KafkaProtocol.decode_produce_response)
+                    partial(KafkaProtocol.encode_produce_request, acks=acks, timeout=timeout),
+                    KafkaProtocol.decode_produce_response)
         out = []
         for resp in resps:
             # Check for errors
diff --git a/kafka/producer.py b/kafka/producer.py
index 93d6e3b..47e690b 100644
--- a/kafka/producer.py
+++ b/kafka/producer.py
@@ -16,8 +16,8 @@ class SimpleProducer(object):
         self.client._load_metadata_for_topics(topic)
         self.next_partition = cycle(self.client.topic_partitions[topic])
 
-    def send_message(self, msg):
+    def send_messages(self, *msg):
         req = ProduceRequest(self.topic, self.next_partition.next(),
-            messages=[create_message(msg)])
+            messages=[create_message(m) for m in msg])
         resp = self.client.send_produce_request([req])[0]
         assert resp.error == 0