code: def encode_produce_request(cls, payloads=(), acks=1, timeout=1000):
          
          if acks not in (1, 0, -1):
              raise ValueError('ProduceRequest acks (%s) must be 1, 0, -1' % acks)
  
          topics = []
          for topic, topic_payloads in group_by_topic_and_partition(payloads).items():
              topic_msgs = []
              for partition, payload in topic_payloads.items():
                  partition_msgs = []
                  for msg in payload.messages:
                      m = kafka.protocol.message.Message(
                            msg.value, key=msg.key,
                            magic=msg.magic, attributes=msg.attributes
                      )
                      partition_msgs.append((0, m.encode()))
                  topic_msgs.append((partition, MessageSet.encode(partition_msgs, prepend_size=False)))
              topics.append((topic, topic_msgs))
  
  
          return kafka.protocol.produce.ProduceRequest[0](
              required_acks=acks,
              timeout=timeout,
              topics=topics
          )
docs: Encode a ProduceRequest struct
  
          Arguments:
              payloads: list of ProduceRequestPayload
              acks: How "acky" you want the request to be
                  1: written to disk by the leader
                  0: immediate response
                  -1: waits for all replicas to be in sync
              timeout: Maximum time (in ms) the server will wait for replica acks.
                  This is _not_ a socket timeout
  
          Returns: ProduceRequest