'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'

@classmethod
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, partition_msgs))
        topics.append((topic, topic_msgs))
    return kafka.protocol.produce.ProduceRequest[0](required_acks=acks, timeout=timeout, topics=topics)
