code: def encode_produce_request(cls, client_id, correlation_id,
                                 payloads=None, acks=1,
                                 timeout=DEFAULT_REPLICAS_ACK_TIMEOUT_MSECS):
          
          if not isinstance(client_id, bytes):
              raise TypeError('client_id={!r} should be bytes'.format(client_id))
          payloads = [] if payloads is None else payloads
          grouped_payloads = group_by_topic_and_partition(payloads)
  
          message = cls._encode_message_header(client_id, correlation_id,
                                               KafkaCodec.PRODUCE_KEY)
  
          message += struct.pack('>hii', acks, timeout, len(grouped_payloads))
  
          for topic, topic_payloads in grouped_payloads.items():
              message += write_short_ascii(topic)
  
              message += struct.pack('>i', len(topic_payloads))
              for partition, payload in topic_payloads.items():
                  msg_set = KafkaCodec._encode_message_set(payload.messages)
                  message += struct.pack('>ii', partition, len(msg_set))
                  message += msg_set
  
          return message
docs: Encode some ProduceRequest structs
  
          :param bytes client_id:
          :param int correlation_id:
          :param list payloads: list of ProduceRequest
          :param int acks:
  
              How "acky" you want the request to be:
  
              0: immediate response
              1: written to disk by the leader
              2+: waits for this many number of replicas to sync
              -1: waits for all replicas to be in sync
  
          :param int timeout:
              Maximum time the server will wait for acks from replicas.  This is
              _not_ a socket timeout.