/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2017 Magnus Edenhill
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "rd.h"
#include "rdkafka_int.h"
#include "rdkafka_msg.h"
#include "rdkafka_msgset.h"
#include "rdkafka_topic.h"
#include "rdkafka_partition.h"
#include "rdkafka_header.h"
#include "rdkafka_lz4.h"

#include "snappy.h"
#include "rdvarint.h"
#include "crc32c.h"


typedef struct rd_kafka_msgset_writer_s {
        rd_kafka_buf_t *msetw_rkbuf;     /* Backing store buffer (refcounted)*/

        int16_t msetw_ApiVersion;        /* ProduceRequest ApiVersion */
        int     msetw_MsgVersion;        /* MsgVersion to construct */
        int     msetw_features;          /* Protocol features to use */
        int     msetw_msgcntmax;         /* Max number of messages to send
                                          * in a batch. */
        size_t  msetw_messages_len;      /* Total size of Messages, without
                                          * MessageSet header */

        size_t  msetw_MessageSetSize;    /* Current MessageSetSize value */
        size_t  msetw_of_MessageSetSize; /* offset of MessageSetSize */
        size_t  msetw_of_start;          /* offset of MessageSet */

        int     msetw_relative_offsets;  /* Bool: use relative offsets */

        /* For MessageSet v2 */
        int     msetw_Attributes;        /* MessageSet Attributes */
        int64_t msetw_MaxTimestamp;      /* Maximum timestamp in batch */
        size_t  msetw_of_CRC;            /* offset of MessageSet.CRC */

        /* First message information */
        struct {
                size_t     of;  /* rkbuf's first message position */
                int64_t    timestamp;
        } msetw_firstmsg;

        rd_kafka_broker_t *msetw_rkb;    /* @warning Not a refcounted
                                          *          reference! */
        rd_kafka_toppar_t *msetw_rktp;   /* @warning Not a refcounted
                                          *          reference! */
} rd_kafka_msgset_writer_t;



/**
 * @brief Select ApiVersion and MsgVersion to use based on broker's
 *        feature compatibility.
 *
 * @locality broker thread
 */
static RD_INLINE void
rd_kafka_msgset_writer_select_MsgVersion (rd_kafka_msgset_writer_t *msetw) {
        rd_kafka_broker_t *rkb = msetw->msetw_rkb;
        int feature;

        if ((feature = rkb->rkb_features & RD_KAFKA_FEATURE_MSGVER2)) {
                msetw->msetw_ApiVersion = 3;
                msetw->msetw_MsgVersion = 2;
                msetw->msetw_features |= feature;
        } else if ((feature = rkb->rkb_features & RD_KAFKA_FEATURE_MSGVER1)) {
                msetw->msetw_ApiVersion = 2;
                msetw->msetw_MsgVersion = 1;
                msetw->msetw_features |= feature;
        } else {
                if ((feature =
                     rkb->rkb_features & RD_KAFKA_FEATURE_THROTTLETIME)) {
                        msetw->msetw_ApiVersion = 1;
                        msetw->msetw_features |= feature;
                } else
                        msetw->msetw_ApiVersion = 0;
                msetw->msetw_MsgVersion = 0;
        }
}


/**
 * @brief Allocate buffer for messageset writer based on a previously set
 *        up \p msetw.
 *
 * Allocate iovecs to hold all headers and messages,
 * and allocate enough space to allow copies of small messages.
 * The allocated size is the minimum of message.max.bytes
 * or queued_bytes + msgcntmax * msg_overhead
 */
static void
rd_kafka_msgset_writer_alloc_buf (rd_kafka_msgset_writer_t *msetw) {
        rd_kafka_t *rk = msetw->msetw_rkb->rkb_rk;
        size_t msg_overhead = 0;
        size_t hdrsize = 0;
        size_t msgsetsize = 0;
        size_t bufsize;

        rd_kafka_assert(NULL, !msetw->msetw_rkbuf);

        /* Calculate worst-case buffer size, produce header size,
         * message size, etc, this isn't critical but avoids unnecesary
         * extra allocations. The buffer will grow as needed if we get
         * this wrong.
         *
         * ProduceRequest headers go in one iovec:
         *  ProduceRequest v0..2:
         *    RequiredAcks + Timeout +
         *    [Topic + [Partition + MessageSetSize]]
         *
         *  ProduceRequest v3:
         *    TransactionalId + RequiredAcks + Timeout +
         *    [Topic + [Partition + MessageSetSize + MessageSet]]
         */

        /*
         * ProduceRequest header sizes
         */
        switch (msetw->msetw_ApiVersion)
        {
        case 3:
                /* Add TransactionalId */
                hdrsize += RD_KAFKAP_STR_SIZE(rk->rk_eos.TransactionalId);
                /* FALLTHRU */
        case 0:
        case 1:
        case 2:
                hdrsize +=
                        /* RequiredAcks + Timeout + TopicCnt */
                        2 + 4 + 4 +
                        /* Topic */
                        RD_KAFKAP_STR_SIZE(msetw->msetw_rktp->
                                           rktp_rkt->rkt_topic) +
                        /* PartitionCnt + Partition + MessageSetSize */
                        4 + 4 + 4;
                msgsetsize += 4; /* MessageSetSize */
                break;

        default:
                RD_NOTREACHED();
        }

        /*
         * MsgVersion specific sizes:
         * - (Worst-case) Message overhead: message fields
         * - MessageSet header size
         */
        switch (msetw->msetw_MsgVersion)
        {
        case 0:
                /* MsgVer0 */
                msg_overhead = RD_KAFKAP_MESSAGE_V0_OVERHEAD;
                break;
        case 1:
                /* MsgVer1 */
                msg_overhead = RD_KAFKAP_MESSAGE_V1_OVERHEAD;
                break;

        case 2:
                /* MsgVer2 uses varints, we calculate for the worst-case. */
                msg_overhead += RD_KAFKAP_MESSAGE_V2_OVERHEAD;

                /* MessageSet header fields */
                msgsetsize +=
                        8 /* BaseOffset */ +
                        4 /* Length */ +
                        4 /* PartitionLeaderEpoch */ +
                        1 /* Magic (MsgVersion) */ +
                        4 /* CRC (CRC32C) */ +
                        2 /* Attributes */ +
                        4 /* LastOffsetDelta */ +
                        8 /* BaseTimestamp */ +
                        8 /* MaxTimestamp */ +
                        8 /* ProducerId */ +
                        2 /* ProducerEpoch */ +
                        4 /* BaseSequence */ +
                        4 /* RecordCount */;
                break;

        default:
                RD_NOTREACHED();
        }

        /*
         * Calculate total buffer size to allocate
         */
        bufsize = hdrsize + msgsetsize;

        /* If copying for small payloads is enabled, allocate enough
         * space for each message to be copied based on this limit.
         */
        if (rk->rk_conf.msg_copy_max_size > 0) {
                size_t queued_bytes = rd_kafka_msgq_size(&msetw->msetw_rktp->
                                                         rktp_xmit_msgq);
                bufsize += RD_MIN(queued_bytes,
                                  (size_t)rk->rk_conf.msg_copy_max_size *
                                  msetw->msetw_msgcntmax);
        }

        /* Add estimed per-message overhead */
        bufsize += msg_overhead * msetw->msetw_msgcntmax;

        /* Cap allocation at message.max.bytes */
        if (bufsize > (size_t)rk->rk_conf.max_msg_size)
                bufsize = (size_t)rk->rk_conf.max_msg_size;

        /*
         * Allocate iovecs to hold all headers and messages,
         * and allocate auxilliery space for message headers, etc.
         */
        msetw->msetw_rkbuf =
                rd_kafka_buf_new_request(msetw->msetw_rkb, RD_KAFKAP_Produce,
                                         msetw->msetw_msgcntmax/2 + 10,
                                         bufsize);

        rd_kafka_buf_ApiVersion_set(msetw->msetw_rkbuf,
                                    msetw->msetw_ApiVersion,
                                    msetw->msetw_features);
}


/**
 * @brief Write the MessageSet header.
 * @remark Must only be called for MsgVersion 2
 */
static void
rd_kafka_msgset_writer_write_MessageSet_v2_header (
        rd_kafka_msgset_writer_t *msetw) {
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        rd_kafka_t *rk = msetw->msetw_rkb->rkb_rk;

        rd_kafka_assert(NULL, msetw->msetw_ApiVersion >= 3);
        rd_kafka_assert(NULL, msetw->msetw_MsgVersion == 2);

        /* BaseOffset (also store the offset to the start of
         * the messageset header fields) */
        msetw->msetw_of_start = rd_kafka_buf_write_i64(rkbuf, 0);

        /* Length: updated later */
        rd_kafka_buf_write_i32(rkbuf, 0);

        /* PartitionLeaderEpoch (KIP-101) */
        rd_kafka_buf_write_i32(rkbuf, 0);

        /* Magic (MsgVersion) */
        rd_kafka_buf_write_i8(rkbuf, msetw->msetw_MsgVersion);

        /* CRC (CRC32C): updated later.
         * CRC needs to be done after the entire messageset+messages has
         * been constructed and the following header fields updated. :(
         * Save the offset for this position. so it can be udpated later. */
        msetw->msetw_of_CRC = rd_kafka_buf_write_i32(rkbuf, 0);

        /* Attributes: updated later */
        rd_kafka_buf_write_i16(rkbuf, 0);

        /* LastOffsetDelta: updated later */
        rd_kafka_buf_write_i32(rkbuf, 0);

        /* BaseTimestamp: updated later */
        rd_kafka_buf_write_i64(rkbuf, 0);

        /* MaxTimestamp: updated later */
        rd_kafka_buf_write_i64(rkbuf, 0);

        /* ProducerId */
        rd_kafka_buf_write_i64(rkbuf, rk->rk_eos.PID);

        /* ProducerEpoch */
        rd_kafka_buf_write_i16(rkbuf, rk->rk_eos.ProducerEpoch);

        /* BaseSequence */
        rd_kafka_buf_write_i32(rkbuf, -1);

        /* RecordCount: udpated later */
        rd_kafka_buf_write_i32(rkbuf, 0);

}


/**
 * @brief Write ProduceRequest headers.
 *        When this function returns the msgset is ready for
 *        writing individual messages.
 *        msetw_MessageSetSize will have been set to the messageset header.
 */
static void
rd_kafka_msgset_writer_write_Produce_header (rd_kafka_msgset_writer_t *msetw) {

        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        rd_kafka_t *rk = msetw->msetw_rkb->rkb_rk;
        rd_kafka_itopic_t *rkt = msetw->msetw_rktp->rktp_rkt;

        /* V3: TransactionalId */
        if (msetw->msetw_ApiVersion == 3)
                rd_kafka_buf_write_kstr(rkbuf, rk->rk_eos.TransactionalId);

        /* RequiredAcks */
        rd_kafka_buf_write_i16(rkbuf, rkt->rkt_conf.required_acks);

        /* Timeout */
        rd_kafka_buf_write_i32(rkbuf, rkt->rkt_conf.request_timeout_ms);

        /* TopicArrayCnt */
        rd_kafka_buf_write_i32(rkbuf, 1);

        /* Insert topic */
        rd_kafka_buf_write_kstr(rkbuf, rkt->rkt_topic);

        /* PartitionArrayCnt */
        rd_kafka_buf_write_i32(rkbuf, 1);

        /* Partition */
        rd_kafka_buf_write_i32(rkbuf, msetw->msetw_rktp->rktp_partition);

        /* MessageSetSize: Will be finalized later*/
        msetw->msetw_of_MessageSetSize = rd_kafka_buf_write_i32(rkbuf, 0);

        if (msetw->msetw_MsgVersion == 2) {
                /* MessageSet v2 header */
                rd_kafka_msgset_writer_write_MessageSet_v2_header(msetw);
                msetw->msetw_MessageSetSize = RD_KAFKAP_MSGSET_V2_SIZE;
        } else {
                /* Older MessageSet */
                msetw->msetw_MessageSetSize = RD_KAFKAP_MSGSET_V0_SIZE;
        }
}


/**
 * @brief Initialize a ProduceRequest MessageSet writer for
 *        the given broker and partition.
 *
 *        A new buffer will be allocated to fit the pending messages in queue.
 *
 * @returns the number of messages to enqueue
 *
 * @remark This currently constructs the entire ProduceRequest, containing
 *         a single outer MessageSet for a single partition.
 *
 * @locality broker thread
 */
static int rd_kafka_msgset_writer_init (rd_kafka_msgset_writer_t *msetw,
                                         rd_kafka_broker_t *rkb,
                                         rd_kafka_toppar_t *rktp) {
        int msgcnt = rktp->rktp_xmit_msgq.rkmq_msg_cnt;

        if (msgcnt == 0)
                return 0;

        memset(msetw, 0, sizeof(*msetw));

        msetw->msetw_rktp = rktp;
        msetw->msetw_rkb = rkb;

        /* Max number of messages to send in a batch,
         * limited by current queue size or configured batch size,
         * whichever is lower. */
        msetw->msetw_msgcntmax = RD_MIN(msgcnt,
                                        rkb->rkb_rk->rk_conf.
                                        batch_num_messages);
        rd_dassert(msetw->msetw_msgcntmax > 0);

        /* Select MsgVersion to use */
        rd_kafka_msgset_writer_select_MsgVersion(msetw);

        /* MsgVersion specific setup. */
        switch (msetw->msetw_MsgVersion)
        {
        case 2:
                msetw->msetw_relative_offsets = 1; /* OffsetDelta */
                break;
        case 1:
                if (rktp->rktp_rkt->rkt_conf.compression_codec)
                        msetw->msetw_relative_offsets = 1;
                break;
        }

        /* Allocate backing buffer */
        rd_kafka_msgset_writer_alloc_buf(msetw);

        /* Construct first part of Produce header + MessageSet header */
        rd_kafka_msgset_writer_write_Produce_header(msetw);

        /* The current buffer position is now where the first message
         * is located.
         * Record the current buffer position so it can be rewound later
         * in case of compression. */
        msetw->msetw_firstmsg.of = rd_buf_write_pos(&msetw->msetw_rkbuf->
                                                    rkbuf_buf);

        return msetw->msetw_msgcntmax;
}



/**
 * @brief Copy or link message payload to buffer.
 */
static RD_INLINE void
rd_kafka_msgset_writer_write_msg_payload (rd_kafka_msgset_writer_t *msetw,
                                          const rd_kafka_msg_t *rkm,
                                          void (*free_cb)(void *)) {
        const rd_kafka_t *rk = msetw->msetw_rkb->rkb_rk;
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;

        /* If payload is below the copy limit and there is still
         * room in the buffer we'll copy the payload to the buffer,
         * otherwise we push a reference to the memory. */
        if (rkm->rkm_len <= (size_t)rk->rk_conf.msg_copy_max_size &&
            rd_buf_write_remains(&rkbuf->rkbuf_buf) > rkm->rkm_len) {
                rd_kafka_buf_write(rkbuf,
                                   rkm->rkm_payload, rkm->rkm_len);
                if (free_cb)
                        free_cb(rkm->rkm_payload);
        } else
                rd_kafka_buf_push(rkbuf, rkm->rkm_payload, rkm->rkm_len,
                                  free_cb);
}


/**
 * @brief Write message headers to buffer.
 *
 * @remark The enveloping HeaderCount varint must already have been written.
 * @returns the number of bytes written to msetw->msetw_rkbuf
 */
static size_t
rd_kafka_msgset_writer_write_msg_headers (rd_kafka_msgset_writer_t *msetw,
                                          const rd_kafka_headers_t *hdrs) {
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        const rd_kafka_header_t *hdr;
        int i;
        size_t start_pos = rd_buf_write_pos(&rkbuf->rkbuf_buf);
        size_t written;

        RD_LIST_FOREACH(hdr, &hdrs->rkhdrs_list, i) {
                rd_kafka_buf_write_varint(rkbuf, hdr->rkhdr_name_size);
                rd_kafka_buf_write(rkbuf,
                                   hdr->rkhdr_name, hdr->rkhdr_name_size);
                rd_kafka_buf_write_varint(rkbuf,
                                          hdr->rkhdr_value ?
                                          (int64_t)hdr->rkhdr_value_size : -1);
                rd_kafka_buf_write(rkbuf,
                                   hdr->rkhdr_value,
                                   hdr->rkhdr_value_size);
        }

        written = rd_buf_write_pos(&rkbuf->rkbuf_buf) - start_pos;
        rd_dassert(written == hdrs->rkhdrs_ser_size);

        return written;
}



/**
 * @brief Write message to messageset buffer with MsgVersion 0 or 1.
 * @returns the number of bytes written.
 */
static size_t
rd_kafka_msgset_writer_write_msg_v0_1 (rd_kafka_msgset_writer_t *msetw,
                                       rd_kafka_msg_t *rkm,
                                       int64_t Offset,
                                       int8_t MsgAttributes,
                                       void (*free_cb)(void *)) {
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        size_t MessageSize;
        size_t of_Crc;

        /*
         * MessageSet's (v0 and v1) per-Message header.
         */

        /* Offset (only relevant for compressed messages on MsgVersion v1) */
        rd_kafka_buf_write_i64(rkbuf, Offset);

        /* MessageSize */
        MessageSize =
                4 + 1 + 1 + /* Crc+MagicByte+Attributes */
                4 /* KeyLength */ + rkm->rkm_key_len +
                4 /* ValueLength */ + rkm->rkm_len;

        if (msetw->msetw_MsgVersion == 1)
                MessageSize += 8; /* Timestamp i64 */

        rd_kafka_buf_write_i32(rkbuf, (int32_t)MessageSize);

        /*
         * Message
         */
        /* Crc: will be updated later */
        of_Crc = rd_kafka_buf_write_i32(rkbuf, 0);

        /* Start Crc calculation of all buf writes. */
        rd_kafka_buf_crc_init(rkbuf);

        /* MagicByte */
        rd_kafka_buf_write_i8(rkbuf, msetw->msetw_MsgVersion);

        /* Attributes */
        rd_kafka_buf_write_i8(rkbuf, MsgAttributes);

        /* V1: Timestamp */
        if (msetw->msetw_MsgVersion == 1)
                rd_kafka_buf_write_i64(rkbuf, rkm->rkm_timestamp);

        /* Message Key */
        rd_kafka_buf_write_bytes(rkbuf, rkm->rkm_key, rkm->rkm_key_len);

        /* Write or copy Value/payload */
        if (rkm->rkm_payload) {
                rd_kafka_buf_write_i32(rkbuf, (int32_t)rkm->rkm_len);
                rd_kafka_msgset_writer_write_msg_payload(msetw, rkm, free_cb);
        } else
                rd_kafka_buf_write_i32(rkbuf, RD_KAFKAP_BYTES_LEN_NULL);

        /* Finalize Crc */
        rd_kafka_buf_update_u32(rkbuf, of_Crc,
                                rd_kafka_buf_crc_finalize(rkbuf));


        /* Return written message size */
        return 8/*Offset*/ + 4/*MessageSize*/ + MessageSize;
}

/**
 * @brief Write message to messageset buffer with MsgVersion 2.
 * @returns the number of bytes written.
 */
static size_t
rd_kafka_msgset_writer_write_msg_v2 (rd_kafka_msgset_writer_t *msetw,
                                     rd_kafka_msg_t *rkm,
                                     int64_t Offset,
                                     int8_t MsgAttributes,
                                     void (*free_cb)(void *)) {
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        size_t MessageSize = 0;
        char varint_Length[RD_UVARINT_ENC_SIZEOF(int32_t)];
        char varint_TimestampDelta[RD_UVARINT_ENC_SIZEOF(int64_t)];
        char varint_OffsetDelta[RD_UVARINT_ENC_SIZEOF(int64_t)];
        char varint_KeyLen[RD_UVARINT_ENC_SIZEOF(int32_t)];
        char varint_ValueLen[RD_UVARINT_ENC_SIZEOF(int32_t)];
        char varint_HeaderCount[RD_UVARINT_ENC_SIZEOF(int32_t)];
        size_t sz_Length;
        size_t sz_TimestampDelta;
        size_t sz_OffsetDelta;
        size_t sz_KeyLen;
        size_t sz_ValueLen;
        size_t sz_HeaderCount;
        int    HeaderCount = 0;
        size_t HeaderSize = 0;

        if (rkm->rkm_headers) {
                HeaderCount = rkm->rkm_headers->rkhdrs_list.rl_cnt;
                HeaderSize  = rkm->rkm_headers->rkhdrs_ser_size;
        }

        /* All varints, except for Length, needs to be pre-built
         * so that the Length field can be set correctly and thus have
         * correct varint encoded width. */

        sz_TimestampDelta = rd_uvarint_enc_i64(
                varint_TimestampDelta, sizeof(varint_TimestampDelta),
                rkm->rkm_timestamp - msetw->msetw_firstmsg.timestamp);
        sz_OffsetDelta = rd_uvarint_enc_i64(
                varint_OffsetDelta, sizeof(varint_OffsetDelta), Offset);
        sz_KeyLen = rd_uvarint_enc_i32(
                varint_KeyLen, sizeof(varint_KeyLen),
                rkm->rkm_key ? (int32_t)rkm->rkm_key_len :
                (int32_t)RD_KAFKAP_BYTES_LEN_NULL);
        sz_ValueLen = rd_uvarint_enc_i32(
                varint_ValueLen, sizeof(varint_ValueLen),
                rkm->rkm_payload ? (int32_t)rkm->rkm_len :
                (int32_t)RD_KAFKAP_BYTES_LEN_NULL);
        sz_HeaderCount = rd_uvarint_enc_i32(
                varint_HeaderCount, sizeof(varint_HeaderCount),
                (int32_t)HeaderCount);

        /* Calculate MessageSize without length of Length (added later)
         * to store it in Length. */
        MessageSize =
                1 /* MsgAttributes */ +
                sz_TimestampDelta +
                sz_OffsetDelta +
                sz_KeyLen +
                rkm->rkm_key_len +
                sz_ValueLen +
                rkm->rkm_len +
                sz_HeaderCount +
                HeaderSize;

        /* Length */
        sz_Length = rd_uvarint_enc_i64(varint_Length, sizeof(varint_Length),
                                       MessageSize);
        rd_kafka_buf_write(rkbuf, varint_Length, sz_Length);
        MessageSize += sz_Length;

        /* Attributes: The MsgAttributes argument is losely based on MsgVer0
         *             which don't apply for MsgVer2 */
        rd_kafka_buf_write_i8(rkbuf, 0);

        /* TimestampDelta */
        rd_kafka_buf_write(rkbuf, varint_TimestampDelta, sz_TimestampDelta);

        /* OffsetDelta */
        rd_kafka_buf_write(rkbuf, varint_OffsetDelta, sz_OffsetDelta);

        /* KeyLen */
        rd_kafka_buf_write(rkbuf, varint_KeyLen, sz_KeyLen);

        /* Key (if any) */
        if (rkm->rkm_key)
                rd_kafka_buf_write(rkbuf, rkm->rkm_key, rkm->rkm_key_len);

        /* ValueLen */
        rd_kafka_buf_write(rkbuf, varint_ValueLen, sz_ValueLen);

        /* Write or copy Value/payload */
        if (rkm->rkm_payload)
                rd_kafka_msgset_writer_write_msg_payload(msetw, rkm, free_cb);

        /* HeaderCount */
        rd_kafka_buf_write(rkbuf, varint_HeaderCount, sz_HeaderCount);

        /* Headers array */
        if (rkm->rkm_headers)
                rd_kafka_msgset_writer_write_msg_headers(msetw,
                                                         rkm->rkm_headers);

        /* Return written message size */
        return MessageSize;
}


/**
 * @brief Write message to messageset buffer.
 * @returns the number of bytes written.
 */
static size_t
rd_kafka_msgset_writer_write_msg (rd_kafka_msgset_writer_t *msetw,
                                  rd_kafka_msg_t *rkm,
                                  int64_t Offset, int8_t MsgAttributes,
                                  void (*free_cb)(void *)) {
        size_t outlen;
        size_t (*writer[]) (rd_kafka_msgset_writer_t *,
                            rd_kafka_msg_t *, int64_t, int8_t,
                            void (*)(void *)) = {
                [0] = rd_kafka_msgset_writer_write_msg_v0_1,
                [1] = rd_kafka_msgset_writer_write_msg_v0_1,
                [2] = rd_kafka_msgset_writer_write_msg_v2
        };
        size_t actual_written;
        size_t pre_pos;

        if (likely(rkm->rkm_timestamp))
                MsgAttributes |= RD_KAFKA_MSG_ATTR_CREATE_TIME;

        pre_pos = rd_buf_write_pos(&msetw->msetw_rkbuf->rkbuf_buf);

        outlen = writer[msetw->msetw_MsgVersion](msetw, rkm,
                                                 Offset, MsgAttributes,
                                                 free_cb);

        actual_written = rd_buf_write_pos(&msetw->msetw_rkbuf->rkbuf_buf) -
                pre_pos;
        rd_assert(outlen <=
                   rd_kafka_msg_wire_size(rkm, msetw->msetw_MsgVersion));
        rd_assert(outlen == actual_written);

        return outlen;

}

/**
 * @brief Write as many messages from the given message queue to
 *        the messageset.
 *
 *        May not write any messages.
 */
static void
rd_kafka_msgset_writer_write_msgq (rd_kafka_msgset_writer_t *msetw,
                                   rd_kafka_msgq_t *rkmq) {
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        rd_kafka_toppar_t *rktp = msetw->msetw_rktp;
        rd_kafka_broker_t *rkb = msetw->msetw_rkb;
        size_t len = rd_buf_len(&msetw->msetw_rkbuf->rkbuf_buf);
        size_t max_msg_size = (size_t)msetw->msetw_rkb->rkb_rk->
                rk_conf.max_msg_size;
        rd_ts_t int_latency_base;
        rd_ts_t MaxTimestamp = 0;
        rd_kafka_msg_t *rkm;
        int msgcnt = 0;
        const rd_ts_t now = rd_clock();

        /* Internal latency calculation base.
         * Uses rkm_ts_timeout which is enqueue time + timeout */
        int_latency_base = now +
                (rktp->rktp_rkt->rkt_conf.message_timeout_ms * 1000);

        /* Acquire BaseTimestamp from first message. */
        rkm = TAILQ_FIRST(&rkmq->rkmq_msgs);
        rd_kafka_assert(NULL, rkm);
        msetw->msetw_firstmsg.timestamp = rkm->rkm_timestamp;

        /*
         * Write as many messages as possible until buffer is full
         * or limit reached.
         */
        do {
                if (unlikely(msgcnt == msetw->msetw_msgcntmax ||
                             len + rd_kafka_msg_wire_size(rkm, msetw->
                                                          msetw_MsgVersion) >
                             max_msg_size)) {
                        rd_rkb_dbg(rkb, MSG, "PRODUCE",
                                   "No more space in current MessageSet "
                                   "(%i message(s), %"PRIusz" bytes)",
                                   msgcnt, len);
                        break;
                }

                if (unlikely(rkm->rkm_u.producer.ts_backoff > now)) {
                        /* Stop accumulation when we've reached
                         * a message with a retry backoff in the future */
                        break;
                }

                /* Move message to buffer's queue */
                rd_kafka_msgq_deq(rkmq, rkm, 1);
                rd_kafka_msgq_enq(&rkbuf->rkbuf_msgq, rkm);

                /* Add internal latency metrics */
                rd_avg_add(&rkb->rkb_avg_int_latency,
                           int_latency_base - rkm->rkm_ts_timeout);

                /* MessageSet v2's .MaxTimestamp field */
                if (unlikely(MaxTimestamp < rkm->rkm_timestamp))
                        MaxTimestamp = rkm->rkm_timestamp;

                /* Write message to buffer */
                len += rd_kafka_msgset_writer_write_msg(msetw, rkm, msgcnt, 0,
                                                        NULL);

                rd_dassert(len <= max_msg_size);
                msgcnt++;

        } while ((rkm = TAILQ_FIRST(&rkmq->rkmq_msgs)));

        msetw->msetw_MaxTimestamp = MaxTimestamp;
}


#if WITH_ZLIB
/**
 * @brief Compress messageset using gzip/zlib
 */
static int
rd_kafka_msgset_writer_compress_gzip (rd_kafka_msgset_writer_t *msetw,
                                      rd_slice_t *slice,
                                      struct iovec *ciov) {

        rd_kafka_broker_t *rkb = msetw->msetw_rkb;
        rd_kafka_toppar_t *rktp = msetw->msetw_rktp;
        z_stream strm;
        size_t len = rd_slice_remains(slice);
        const void *p;
        size_t rlen;
        int r;

        memset(&strm, 0, sizeof(strm));
        r = deflateInit2(&strm, Z_DEFAULT_COMPRESSION,
                         Z_DEFLATED, 15+16,
                         8, Z_DEFAULT_STRATEGY);
        if (r != Z_OK) {
                rd_rkb_log(rkb, LOG_ERR, "GZIP",
                           "Failed to initialize gzip for "
                           "compressing %"PRIusz" bytes in "
                           "topic %.*s [%"PRId32"]: %s (%i): "
                           "sending uncompressed",
                           len,
                           RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                           rktp->rktp_partition,
                           strm.msg ? strm.msg : "", r);
                return -1;
        }

        /* Calculate maximum compressed size and
         * allocate an output buffer accordingly, being
         * prefixed with the Message header. */
        ciov->iov_len = deflateBound(&strm, (uLong)rd_slice_remains(slice));
        ciov->iov_base = rd_malloc(ciov->iov_len);

        strm.next_out  = (void *)ciov->iov_base;
        strm.avail_out =   (uInt)ciov->iov_len;

        /* Iterate through each segment and compress it. */
        while ((rlen = rd_slice_reader(slice, &p))) {

                strm.next_in  = (void *)p;
                strm.avail_in =   (uInt)rlen;

                /* Compress message */
                if ((r = deflate(&strm, Z_NO_FLUSH) != Z_OK)) {
                        rd_rkb_log(rkb, LOG_ERR, "GZIP",
                                   "Failed to gzip-compress "
                                   "%"PRIusz" bytes (%"PRIusz" total) for "
                                   "topic %.*s [%"PRId32"]: "
                                   "%s (%i): "
                                   "sending uncompressed",
                                   rlen, len,
                                   RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                                   rktp->rktp_partition,
                                   strm.msg ? strm.msg : "", r);
                        deflateEnd(&strm);
                        rd_free(ciov->iov_base);
                        return -1;
                }

                rd_kafka_assert(rkb->rkb_rk, strm.avail_in == 0);
        }

        /* Finish the compression */
        if ((r = deflate(&strm, Z_FINISH)) != Z_STREAM_END) {
                rd_rkb_log(rkb, LOG_ERR, "GZIP",
                           "Failed to finish gzip compression "
                           " of %"PRIusz" bytes for "
                           "topic %.*s [%"PRId32"]: "
                           "%s (%i): "
                           "sending uncompressed",
                           len,
                           RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                           rktp->rktp_partition,
                           strm.msg ? strm.msg : "", r);
                deflateEnd(&strm);
                rd_free(ciov->iov_base);
                return -1;
        }

        ciov->iov_len = strm.total_out;

        /* Deinitialize compression */
        deflateEnd(&strm);

        return 0;
}
#endif


#if WITH_SNAPPY
/**
 * @brief Compress messageset using Snappy
 */
static int
rd_kafka_msgset_writer_compress_snappy (rd_kafka_msgset_writer_t *msetw,
                                        rd_slice_t *slice, struct iovec *ciov) {
        rd_kafka_broker_t *rkb = msetw->msetw_rkb;
        rd_kafka_toppar_t *rktp = msetw->msetw_rktp;
        struct iovec *iov;
        size_t iov_max, iov_cnt;
        struct snappy_env senv;
        size_t len = rd_slice_remains(slice);
        int r;

        /* Initialize snappy compression environment */
        rd_kafka_snappy_init_env_sg(&senv, 1/*iov enable*/);

        /* Calculate maximum compressed size and
         * allocate an output buffer accordingly. */
        ciov->iov_len = rd_kafka_snappy_max_compressed_length(len);
        ciov->iov_base = rd_malloc(ciov->iov_len);

        iov_max = slice->buf->rbuf_segment_cnt;
        iov = rd_alloca(sizeof(*iov) * iov_max);

        rd_slice_get_iov(slice, iov, &iov_cnt, iov_max, len);

        /* Compress each message */
        if ((r = rd_kafka_snappy_compress_iov(&senv, iov, iov_cnt, len,
                                              ciov)) != 0) {
                rd_rkb_log(rkb, LOG_ERR, "SNAPPY",
                           "Failed to snappy-compress "
                           "%"PRIusz" bytes for "
                           "topic %.*s [%"PRId32"]: %s: "
                           "sending uncompressed",
                           len, RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                           rktp->rktp_partition,
                           rd_strerror(-r));
                rd_free(ciov->iov_base);
                return -1;
        }

        /* rd_free snappy environment */
        rd_kafka_snappy_free_env(&senv);

        return 0;
}
#endif

/**
 * @brief Compress messageset using LZ4F
 */
static int
rd_kafka_msgset_writer_compress_lz4 (rd_kafka_msgset_writer_t *msetw,
                                     rd_slice_t *slice, struct iovec *ciov) {
        rd_kafka_resp_err_t err;
        err = rd_kafka_lz4_compress(msetw->msetw_rkb,
                                    /* Correct or incorrect HC */
                                    msetw->msetw_MsgVersion >= 1 ? 1 : 0,
                                    slice, &ciov->iov_base, &ciov->iov_len);
        return (err ? -1 : 0);
}



/**
 * @brief Compress the message set.
 * @param outlenp in: total uncompressed messages size,
 *                out (on success): returns the compressed buffer size.
 * @returns 0 on success or if -1 if compression failed.
 * @remark Compression failures are not critical, we'll just send the
 *         the messageset uncompressed.
 */
static int
rd_kafka_msgset_writer_compress (rd_kafka_msgset_writer_t *msetw,
                                 size_t *outlenp) {
        rd_kafka_toppar_t *rktp = msetw->msetw_rktp;
        rd_buf_t *rbuf = &msetw->msetw_rkbuf->rkbuf_buf;
        rd_slice_t slice;
        size_t len = *outlenp;
        struct iovec ciov = RD_ZERO_INIT; /* Compressed output buffer */
        int r = -1;
        size_t outlen;

        rd_assert(rd_buf_len(rbuf) >= msetw->msetw_firstmsg.of + len);

        /* Create buffer slice from firstmsg and onwards */
        r = rd_slice_init(&slice, rbuf, msetw->msetw_firstmsg.of, len);
        rd_assert(r == 0 || !*"invalid firstmsg position");

        switch (rktp->rktp_rkt->rkt_conf.compression_codec)
        {
#if WITH_ZLIB
        case RD_KAFKA_COMPRESSION_GZIP:
                r = rd_kafka_msgset_writer_compress_gzip(msetw, &slice, &ciov);
                break;
#endif

#if WITH_SNAPPY
        case RD_KAFKA_COMPRESSION_SNAPPY:
                r = rd_kafka_msgset_writer_compress_snappy(msetw, &slice,
                                                           &ciov);
                break;
#endif

        case RD_KAFKA_COMPRESSION_LZ4:
                /* Skip LZ4 compression if broker doesn't support it. */
                if (!(msetw->msetw_rkb->rkb_features & RD_KAFKA_FEATURE_LZ4))
                        return -1;

                r = rd_kafka_msgset_writer_compress_lz4(msetw, &slice, &ciov);
                break;


        default:
                rd_kafka_assert(NULL,
                                !*"notreached: unsupported compression.codec");
                break;
        }

        if (r == -1) /* Compression failed, send uncompressed */
                return -1;


        if (unlikely(ciov.iov_len > len)) {
                /* If the compressed data is larger than the uncompressed size
                 * then throw it away and send as uncompressed. */
                rd_free(ciov.iov_base);
                return -1;
        }

        /* Set compression codec in MessageSet.Attributes */
        msetw->msetw_Attributes |= rktp->rktp_rkt->rkt_conf.compression_codec;

        /* Rewind rkbuf to the pre-message checkpoint (firstmsg)
         * and replace the original message(s) with the compressed payload,
         * possibly with version dependent enveloping. */
        rd_buf_write_seek(rbuf, msetw->msetw_firstmsg.of);

        rd_kafka_assert(msetw->msetw_rkb->rkb_rk, ciov.iov_len < INT32_MAX);

        if (msetw->msetw_MsgVersion == 2) {
                /* MsgVersion 2 has no inner MessageSet header or wrapping
                 * for compressed messages, just the messages back-to-back,
                 * so we can push the compressed memory directly to the
                 * buffer without wrapping it. */
                rd_buf_push(rbuf, ciov.iov_base, ciov.iov_len, rd_free);
                outlen = ciov.iov_len;

        } else {
                /* Older MessageSets envelope/wrap the compressed MessageSet
                 * in an outer Message. */
                rd_kafka_msg_t rkm = {
                        .rkm_len       = ciov.iov_len,
                        .rkm_payload   = ciov.iov_base,
                        .rkm_timestamp = msetw->msetw_firstmsg.timestamp
                };
                outlen = rd_kafka_msgset_writer_write_msg(
                        msetw, &rkm, 0,
                        rktp->rktp_rkt->rkt_conf.compression_codec,
                        rd_free/*free for ciov.iov_base*/);
        }

        *outlenp = outlen;

        return 0;
}




/**
 * @brief Calculate MessageSet v2 CRC (CRC32C) when messageset is complete.
 */
static void
rd_kafka_msgset_writer_calc_crc_v2 (rd_kafka_msgset_writer_t *msetw) {
        int32_t crc;
        rd_slice_t slice;
        int r;

        r = rd_slice_init(&slice, &msetw->msetw_rkbuf->rkbuf_buf,
                          msetw->msetw_of_CRC+4,
                          rd_buf_write_pos(&msetw->msetw_rkbuf->rkbuf_buf) -
                          msetw->msetw_of_CRC-4);
       rd_assert(!r && *"slice_init failed");

       /* CRC32C calculation */
        crc = rd_slice_crc32c(&slice);

        /* Update CRC at MessageSet v2 CRC offset */
        rd_kafka_buf_update_i32(msetw->msetw_rkbuf, msetw->msetw_of_CRC, crc);
}

/**
 * @brief Finalize MessageSet v2 header fields.
 */
static void
rd_kafka_msgset_writer_finalize_MessageSet_v2_header (
        rd_kafka_msgset_writer_t *msetw) {
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        int msgcnt = rd_kafka_msgq_len(&rkbuf->rkbuf_msgq);

        rd_kafka_assert(NULL, msgcnt > 0);
        rd_kafka_assert(NULL, msetw->msetw_ApiVersion >= 3);

        msetw->msetw_MessageSetSize = RD_KAFKAP_MSGSET_V2_SIZE +
                msetw->msetw_messages_len;

        /* MessageSet.Length is the same as
         * MessageSetSize minus field widths for FirstOffset+Length */
        rd_kafka_buf_update_i32(rkbuf, msetw->msetw_of_start +
                                RD_KAFKAP_MSGSET_V2_OF_Length,
                                (int32_t)msetw->msetw_MessageSetSize - (8+4));

        msetw->msetw_Attributes |= RD_KAFKA_MSG_ATTR_CREATE_TIME;

        rd_kafka_buf_update_i16(rkbuf, msetw->msetw_of_start +
                                RD_KAFKAP_MSGSET_V2_OF_Attributes,
                                msetw->msetw_Attributes);

        rd_kafka_buf_update_i32(rkbuf, msetw->msetw_of_start +
                                RD_KAFKAP_MSGSET_V2_OF_LastOffsetDelta,
                                msgcnt-1);

        rd_kafka_buf_update_i64(rkbuf, msetw->msetw_of_start +
                                RD_KAFKAP_MSGSET_V2_OF_BaseTimestamp,
                                msetw->msetw_firstmsg.timestamp);

        rd_kafka_buf_update_i64(rkbuf, msetw->msetw_of_start +
                                RD_KAFKAP_MSGSET_V2_OF_MaxTimestamp,
                                msetw->msetw_MaxTimestamp);

        rd_kafka_buf_update_i32(rkbuf, msetw->msetw_of_start +
                                RD_KAFKAP_MSGSET_V2_OF_RecordCount, msgcnt);

        rd_kafka_msgset_writer_calc_crc_v2(msetw);
}




/**
 * @brief Finalize the MessageSet header, if applicable.
 */
static void
rd_kafka_msgset_writer_finalize_MessageSet (rd_kafka_msgset_writer_t *msetw) {
        rd_dassert(msetw->msetw_messages_len > 0);

        if (msetw->msetw_MsgVersion == 2)
                rd_kafka_msgset_writer_finalize_MessageSet_v2_header(msetw);
        else
                msetw->msetw_MessageSetSize = RD_KAFKAP_MSGSET_V0_SIZE +
                        msetw->msetw_messages_len;

        /* Update MessageSetSize */
        rd_kafka_buf_update_i32(msetw->msetw_rkbuf,
                                msetw->msetw_of_MessageSetSize,
                                (int32_t)msetw->msetw_MessageSetSize);

}


/**
 * @brief Finalize the messageset - call when no more messages are to be
 *        added to the messageset.
 *
 *        Will compress, update final values, CRCs, etc.
 *
 *        The messageset writer is destroyed and the buffer is returned
 *        and ready to be transmitted.
 *
 * @param MessagetSetSizep will be set to the finalized MessageSetSize
 *
 * @returns the buffer to transmit or NULL if there were no messages
 *          in messageset.
 */
static rd_kafka_buf_t *
rd_kafka_msgset_writer_finalize (rd_kafka_msgset_writer_t *msetw,
                                 size_t *MessageSetSizep) {
        rd_kafka_buf_t *rkbuf = msetw->msetw_rkbuf;
        rd_kafka_toppar_t *rktp = msetw->msetw_rktp;
        size_t len;
        int cnt;

        /* No messages added, bail out early. */
        if (unlikely((cnt = rd_kafka_msgq_len(&rkbuf->rkbuf_msgq)) == 0)) {
                rd_kafka_buf_destroy(rkbuf);
                return NULL;
        }

        /* Total size of messages */
        len = rd_buf_write_pos(&msetw->msetw_rkbuf->rkbuf_buf) -
                msetw->msetw_firstmsg.of;
        rd_assert(len > 0);
        rd_assert(len <= (size_t)rktp->rktp_rkt->rkt_rk->rk_conf.max_msg_size);

        /* Compress the message set */
        if (rktp->rktp_rkt->rkt_conf.compression_codec)
                rd_kafka_msgset_writer_compress(msetw, &len);

        msetw->msetw_messages_len = len;

        /* Finalize MessageSet header fields */
        rd_kafka_msgset_writer_finalize_MessageSet(msetw);

        /* Return final MessageSetSize */
        *MessageSetSizep = msetw->msetw_MessageSetSize;

        rd_rkb_dbg(msetw->msetw_rkb, MSG, "PRODUCE",
                   "%s [%"PRId32"]: "
                   "Produce MessageSet with %i message(s) (%"PRIusz" bytes, "
                   "ApiVersion %d, MsgVersion %d)",
                   rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition,
                   cnt, msetw->msetw_MessageSetSize,
                   msetw->msetw_ApiVersion, msetw->msetw_MsgVersion);


        return rkbuf;
}


/**
 * @brief Create ProduceRequest containing as many messages from
 *        the toppar's transmit queue as possible, limited by configuration,
 *        size, etc.
 *
 * @param rkb broker to create buffer for
 * @param rktp toppar to transmit messages for
 * @param MessagetSetSizep will be set to the final MessageSetSize
 *
 * @returns the buffer to transmit or NULL if there were no messages
 *          in messageset.
 *
 * @locality broker thread
 */
rd_kafka_buf_t *
rd_kafka_msgset_create_ProduceRequest (rd_kafka_broker_t *rkb,
                                       rd_kafka_toppar_t *rktp,
                                       size_t *MessageSetSizep) {

        rd_kafka_msgset_writer_t msetw;

        if (rd_kafka_msgset_writer_init(&msetw, rkb, rktp) == 0)
                return NULL;

        rd_kafka_msgset_writer_write_msgq(&msetw, &rktp->rktp_xmit_msgq);

        return rd_kafka_msgset_writer_finalize(&msetw, MessageSetSizep);
}
