'Encode and send some ProduceRequests
ProduceRequests will be grouped by (topic, partition) and then
sent to a specific broker. Output is a list of responses in the
same order as the list of payloads specified
Arguments:
payloads (list of ProduceRequest): produce requests to send to kafka
ProduceRequest payloads must not contain duplicates for any
topic-partition.
acks (int, optional): how many acks the servers should receive from replica
brokers before responding to the request. If it is 0, the server
will not send any response. If it is 1, the server will wait
until the data is written to the local log before sending a
response.  If it is -1, the server will wait until the message
is committed by all in-sync replicas before sending a response.
For any value > 1, the server will wait for this number of acks to
occur (but the server will never wait for more acknowledgements than
there are in-sync replicas). defaults to 1.
timeout (int, optional): maximum time in milliseconds the server can
await the receipt of the number of acks, defaults to 1000.
fail_on_error (bool, optional): raise exceptions on connection and
server response errors, defaults to True.
callback (function, optional): instead of returning the ProduceResponse,
first pass it through this function, defaults to None.
Returns:
list of ProduceResponses, or callback results if supplied, in the
order of input payloads'

def send_produce_request(self, payloads=[], acks=1, timeout=1000, fail_on_error=True, callback=None):

    encoder = functools.partial(KafkaProtocol.encode_produce_request, acks=acks, timeout=timeout)
    if (acks == 0):
        decoder = None
    else:
        decoder = KafkaProtocol.decode_produce_response
    resps = self._send_broker_aware_request(payloads, encoder, decoder)
    return [(resp if (not callback) else callback(resp)) for resp in resps if ((resp is not None) and ((not fail_on_error) or (not self._raise_on_response_error(resp))))]
