@classmethod<EOL><INDENT>def encode_produce_request(cls, client_id, correlation_id,<EOL>payloads=None, acks=<NUM_LIT:1>, timeout=<NUM_LIT:1000>):<DEDENT>
payloads = [] if payloads is None else payloads<EOL>grouped_payloads = group_by_topic_and_partition(payloads)<EOL>message = []<EOL>message.append(cls._encode_message_header(client_id, correlation_id,<EOL>KafkaProtocol.PRODUCE_KEY))<EOL>message.append(struct.pack('<STR_LIT>', acks, timeout,<EOL>len(grouped_payloads)))<EOL>for topic, topic_payloads in grouped_payloads.items():<EOL><INDENT>message.append(struct.pack('<STR_LIT>' % len(topic), len(topic), topic,<EOL>len(topic_payloads)))<EOL>for partition, payload in topic_payloads.items():<EOL><INDENT>msg_set = KafkaProtocol._encode_message_set(payload.messages)<EOL>message.append(struct.pack('<STR_LIT>' % len(msg_set), partition,<EOL>len(msg_set), msg_set))<EOL><DEDENT><DEDENT>msg = b'<STR_LIT>'.join(message)<EOL>return struct.pack('<STR_LIT>' % len(msg), len(msg), msg)<EOL>
Encode some ProduceRequest structs

Arguments:
    client_id: string
    correlation_id: int
    payloads: list of ProduceRequest
    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
    timeout: Maximum time the server will wait for acks from replicas.
        This is _not_ a socket timeout