code: def send_produce_request(self, payloads=None, acks=1,
                               timeout=DEFAULT_REPLICAS_ACK_MSECS,
                               fail_on_error=True, callback=None):
          
  
          encoder = partial(
              KafkaCodec.encode_produce_request,
              acks=acks,
              timeout=timeout)
  
          if acks == 0:
              decoder = None
          else:
              decoder = KafkaCodec.decode_produce_response
  
          resps = yield self._send_broker_aware_request(
              payloads, encoder, decoder)
  
          returnValue(self._handle_responses(resps, fail_on_error, callback))
docs: 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
  
          Parameters
          ----------
          payloads:
              list of ProduceRequest
          acks:
              How many Kafka broker replicas need to write before
              the leader replies with a response
          timeout:
              How long the server has to receive the acks from the
              replicas before returning an error.
          fail_on_error:
              boolean, should we raise an Exception if we encounter an API error?
          callback:
              function, instead of returning the ProduceResponse,
              first pass it through this function
  
          Return
          ------
          a deferred which callbacks with a list of ProduceResponse
  
          Raises
          ------
          FailedPayloadsError, LeaderUnavailableError, PartitionUnavailableError