http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_msgset_reader.c
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_msgset_reader.c
b/thirdparty/librdkafka-0.11.1/src/rdkafka_msgset_reader.c
deleted file mode 100644
index a073819..0000000
--- a/thirdparty/librdkafka-0.11.1/src/rdkafka_msgset_reader.c
+++ /dev/null
@@ -1,1090 +0,0 @@
-/*
- * 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.
- */
-
-/**
- * @name MessageSet reader interface
- *
- * Parses FetchResponse for Messages
- *
- *
- * @remark
- * The broker may send partial messages, when this happens we bail out
- * silently and keep the messages that we successfully parsed.
- *
- * "A Guide To The Kafka Protocol" states:
- * "As an optimization the server is allowed to
- * return a partial message at the end of the
- * message set.
- * Clients should handle this case."
- *
- * We're handling it by not passing the error upstream.
- * This is why most err_parse: goto labels (that are called from buf parsing
- * macros) suppress the error message and why log_decode_errors is off
- * unless PROTOCOL debugging is enabled.
- *
- * When a FetchResponse contains multiple partitions, each partition's
- * MessageSet may be partial, regardless of the other partitions.
- * To make sure the next partition can be parsed, each partition parse
- * uses its own sub-slice of only that partition's MessageSetSize length.
- */
-
-#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_lz4.h"
-
-#include "rdvarint.h"
-#include "crc32c.h"
-
-#if WITH_ZLIB
-#include "rdgz.h"
-#endif
-#if WITH_SNAPPY
-#include "snappy.h"
-#endif
-
-
-
-struct msgset_v2_hdr {
- int64_t BaseOffset;
- int32_t Length;
- int32_t PartitionLeaderEpoch;
- int8_t MagicByte;
- int32_t Crc;
- int16_t Attributes;
- int32_t LastOffsetDelta;
- int64_t BaseTimestamp;
- int64_t MaxTimestamp;
- int64_t PID;
- int16_t ProducerEpoch;
- int32_t BaseSequence;
- int32_t RecordCount;
-};
-
-
-typedef struct rd_kafka_msgset_reader_s {
- rd_kafka_buf_t *msetr_rkbuf; /**< Response read buffer */
-
- int msetr_relative_offsets; /**< Bool: using relative offsets */
-
- /**< Outer/wrapper Message fields. */
- struct {
- int64_t offset; /**< Relative_offsets: outer message's
- * Offset (last offset) */
- rd_kafka_timestamp_type_t tstype; /**< Compressed
- * MessageSet's
- * timestamp type. */
- int64_t timestamp; /**< ... timestamp*/
- } msetr_outer;
-
- struct msgset_v2_hdr *msetr_v2_hdr; /**< MessageSet v2 header */
-
- const struct rd_kafka_toppar_ver *msetr_tver; /**< Toppar op version of
- * request. */
-
- rd_kafka_broker_t *msetr_rkb; /* @warning Not a refcounted
- * reference! */
- rd_kafka_toppar_t *msetr_rktp; /* @warning Not a refcounted
- * reference! */
-
- int msetr_msgcnt; /**< Number of messages in rkq */
- rd_kafka_q_t msetr_rkq; /**< Temp Message and error queue */
- rd_kafka_q_t *msetr_par_rkq; /**< Parent message and error queue,
- * the temp msetr_rkq will be moved
- * to this queue when parsing
- * is done.
- * Refcount is not increased. */
-} rd_kafka_msgset_reader_t;
-
-
-
-/* Forward declarations */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_run (rd_kafka_msgset_reader_t *msetr);
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_msgs_v2 (rd_kafka_msgset_reader_t *msetr);
-
-
-/**
- * @brief Set up a MessageSet reader but don't start reading messages.
- */
-static void
-rd_kafka_msgset_reader_init (rd_kafka_msgset_reader_t *msetr,
- rd_kafka_buf_t *rkbuf,
- rd_kafka_toppar_t *rktp,
- const struct rd_kafka_toppar_ver *tver,
- rd_kafka_q_t *par_rkq) {
-
- memset(msetr, 0, sizeof(*msetr));
-
- msetr->msetr_rkb = rkbuf->rkbuf_rkb;
- msetr->msetr_rktp = rktp;
- msetr->msetr_tver = tver;
- msetr->msetr_rkbuf = rkbuf;
-
- /* All parsed messages are put on this temporary op
- * queue first and then moved in one go to the real op queue. */
- rd_kafka_q_init(&msetr->msetr_rkq, msetr->msetr_rkb->rkb_rk);
-
- /* Make sure enqueued ops get the correct serve/opaque reflecting the
- * original queue. */
- msetr->msetr_rkq.rkq_serve = par_rkq->rkq_serve;
- msetr->msetr_rkq.rkq_opaque = par_rkq->rkq_opaque;
-
- /* Keep (non-refcounted) reference to parent queue for
- * moving the messages and events in msetr_rkq to when
- * parsing is done. */
- msetr->msetr_par_rkq = par_rkq;
-}
-
-
-
-
-
-/**
- * @brief Decompress MessageSet, pass the uncompressed MessageSet to
- * the MessageSet reader.
- */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_decompress (rd_kafka_msgset_reader_t *msetr,
- int MsgVersion, int Attributes,
- int64_t Timestamp, int64_t Offset,
- const void *compressed,
- size_t compressed_size) {
- struct iovec iov = { .iov_base = NULL, .iov_len = 0 };
- rd_kafka_toppar_t *rktp = msetr->msetr_rktp;
- int codec = Attributes & RD_KAFKA_MSG_ATTR_COMPRESSION_MASK;
- rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;
- rd_kafka_buf_t *rkbufz;
-
- switch (codec)
- {
-#if WITH_ZLIB
- case RD_KAFKA_COMPRESSION_GZIP:
- {
- uint64_t outlenx = 0;
-
- /* Decompress Message payload */
- iov.iov_base = rd_gz_decompress(compressed,
(int)compressed_size,
- &outlenx);
- if (unlikely(!iov.iov_base)) {
- rd_rkb_dbg(msetr->msetr_rkb, MSG, "GZIP",
- "Failed to decompress Gzip "
- "message at offset %"PRId64
- " of %"PRIusz" bytes: "
- "ignoring message",
- Offset, compressed_size);
- err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
- goto err;
- }
-
- iov.iov_len = (size_t)outlenx;
- }
- break;
-#endif
-
-#if WITH_SNAPPY
- case RD_KAFKA_COMPRESSION_SNAPPY:
- {
- const char *inbuf = compressed;
- size_t inlen = compressed_size;
- int r;
- static const unsigned char snappy_java_magic[] =
- { 0x82, 'S','N','A','P','P','Y', 0 };
- static const size_t snappy_java_hdrlen = 8+4+4;
-
- /* snappy-java adds its own header (SnappyCodec)
- * which is not compatible with the official Snappy
- * implementation.
- * 8: magic, 4: version, 4: compatible
- * followed by any number of chunks:
- * 4: length
- * ...: snappy-compressed data. */
- if (likely(inlen > snappy_java_hdrlen + 4 &&
- !memcmp(inbuf, snappy_java_magic, 8))) {
- /* snappy-java framing */
- char errstr[128];
-
- inbuf = inbuf + snappy_java_hdrlen;
- inlen -= snappy_java_hdrlen;
- iov.iov_base = rd_kafka_snappy_java_uncompress(
- inbuf, inlen,
- &iov.iov_len,
- errstr, sizeof(errstr));
-
- if (unlikely(!iov.iov_base)) {
- rd_rkb_dbg(msetr->msetr_rkb, MSG, "SNAPPY",
- "%s [%"PRId32"]: "
- "Snappy decompression for message "
- "at offset %"PRId64" failed: %s: "
- "ignoring message",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- Offset, errstr);
- err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
- goto err;
- }
-
-
- } else {
- /* No framing */
-
- /* Acquire uncompressed length */
- if (unlikely(!rd_kafka_snappy_uncompressed_length(
- inbuf, inlen, &iov.iov_len))) {
- rd_rkb_dbg(msetr->msetr_rkb, MSG, "SNAPPY",
- "Failed to get length of Snappy "
- "compressed payload "
- "for message at offset %"PRId64
- " (%"PRIusz" bytes): "
- "ignoring message",
- Offset, inlen);
- err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
- goto err;
- }
-
- /* Allocate output buffer for uncompressed data */
- iov.iov_base = rd_malloc(iov.iov_len);
- if (unlikely(!iov.iov_base)) {
- rd_rkb_dbg(msetr->msetr_rkb, MSG, "SNAPPY",
- "Failed to allocate Snappy "
- "decompress buffer of size %"PRIusz
- "for message at offset %"PRId64
- " (%"PRIusz" bytes): %s: "
- "ignoring message",
- iov.iov_len, Offset, inlen,
- rd_strerror(errno));
- err = RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE;
- goto err;
- }
-
- /* Uncompress to outbuf */
- if (unlikely((r = rd_kafka_snappy_uncompress(
- inbuf, inlen, iov.iov_base)))) {
- rd_rkb_dbg(msetr->msetr_rkb, MSG, "SNAPPY",
- "Failed to decompress Snappy "
- "payload for message at offset "
- "%"PRId64" (%"PRIusz" bytes): %s: "
- "ignoring message",
- Offset, inlen,
- rd_strerror(-r/*negative errno*/));
- rd_free(iov.iov_base);
- err = RD_KAFKA_RESP_ERR__BAD_COMPRESSION;
- goto err;
- }
- }
-
- }
- break;
-#endif
-
- case RD_KAFKA_COMPRESSION_LZ4:
- {
- err = rd_kafka_lz4_decompress(msetr->msetr_rkb,
- /* Proper HC? */
- MsgVersion >= 1 ? 1 : 0,
- Offset,
- /* @warning Will modify
compressed
- * if no proper HC */
- (char *)compressed,
- compressed_size,
- &iov.iov_base, &iov.iov_len);
- if (err)
- goto err;
- }
- break;
-
- default:
- rd_rkb_dbg(msetr->msetr_rkb, MSG, "CODEC",
- "%s [%"PRId32"]: Message at offset %"PRId64
- " with unsupported "
- "compression codec 0x%x: message ignored",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- Offset, (int)codec);
-
- err = RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED;
- goto err;
- }
-
-
- rd_assert(iov.iov_base);
-
- /*
- * Decompression successful
- */
-
- /* Create a new buffer pointing to the uncompressed
- * allocated buffer (outbuf) and let messages keep a reference to
- * this new buffer. */
- rkbufz = rd_kafka_buf_new_shadow(iov.iov_base, iov.iov_len, rd_free);
- rkbufz->rkbuf_rkb = msetr->msetr_rkbuf->rkbuf_rkb;
- rd_kafka_broker_keep(rkbufz->rkbuf_rkb);
-
-
- /* In MsgVersion v0..1 the decompressed data contains
- * an inner MessageSet, pass it to a new MessageSet reader.
- *
- * For MsgVersion v2 the decompressed data are the list of messages.
- */
-
- if (MsgVersion <= 1) {
- /* Pass decompressed data (inner Messageset)
- * to new instance of the MessageSet parser. */
- rd_kafka_msgset_reader_t inner_msetr;
- rd_kafka_msgset_reader_init(&inner_msetr,
- rkbufz,
- msetr->msetr_rktp,
- msetr->msetr_tver,
- &msetr->msetr_rkq);
-
- if (MsgVersion == 1) {
- /* postproc() will convert relative to
- * absolute offsets */
- inner_msetr.msetr_relative_offsets = 1;
- inner_msetr.msetr_outer.offset = Offset;
-
- /* Apply single LogAppendTime timestamp for
- * all messages. */
- if (Attributes & RD_KAFKA_MSG_ATTR_LOG_APPEND_TIME) {
- inner_msetr.msetr_outer.tstype =
- RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME;
- inner_msetr.msetr_outer.timestamp = Timestamp;
- }
- }
-
- /* Parse the inner MessageSet */
- err = rd_kafka_msgset_reader_run(&inner_msetr);
-
-
- } else {
- /* MsgVersion 2 */
- rd_kafka_buf_t *orig_rkbuf = msetr->msetr_rkbuf;
-
- /* Temporarily replace read buffer with uncompressed buffer */
- msetr->msetr_rkbuf = rkbufz;
-
- /* Read messages */
- err = rd_kafka_msgset_reader_msgs_v2(msetr);
-
- /* Restore original buffer */
- msetr->msetr_rkbuf = orig_rkbuf;
- }
-
- /* Loose our refcnt of the uncompressed rkbuf.
- * Individual messages/rko's will have their own reference. */
- rd_kafka_buf_destroy(rkbufz);
-
- return err;
-
- err:
- /* Enqueue error messsage:
- * Create op and push on temporary queue. */
- rd_kafka_q_op_err(&msetr->msetr_rkq, RD_KAFKA_OP_CONSUMER_ERR,
- err, msetr->msetr_tver->version, rktp, Offset,
- "Decompression (codec 0x%x) of message at %"PRIu64
- " of %"PRIu64" bytes failed: %s",
- codec, Offset, compressed_size,
rd_kafka_err2str(err));
-
- return err;
-
-}
-
-
-
-/**
- * @brief Message parser for MsgVersion v0..1
- *
- * @returns RD_KAFKA_RESP_ERR_NO_ERROR on success or on single-message errors,
- * or any other error code when the MessageSet parser should stop
- * parsing (such as for partial Messages).
- */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_msg_v0_1 (rd_kafka_msgset_reader_t *msetr) {
- rd_kafka_buf_t *rkbuf = msetr->msetr_rkbuf;
- rd_kafka_toppar_t *rktp = msetr->msetr_rktp;
- rd_kafka_broker_t *rkb = msetr->msetr_rkb;
- struct {
- int64_t Offset; /* MessageSet header */
- int32_t MessageSize; /* MessageSet header */
- uint32_t Crc;
- int8_t MagicByte; /* MsgVersion */
- int8_t Attributes;
- int64_t Timestamp; /* v1 */
- } hdr; /* Message header */
- rd_kafkap_bytes_t Key;
- rd_kafkap_bytes_t Value;
- int32_t Value_len;
- rd_kafka_op_t *rko;
- size_t hdrsize = 6; /* Header size following MessageSize */
- rd_slice_t crc_slice;
- rd_kafka_msg_t *rkm;
- int relative_offsets = 0;
- const char *reloff_str = "";
- /* Only log decoding errors if protocol debugging enabled. */
- int log_decode_errors = (rkbuf->rkbuf_rkb->rkb_rk->rk_conf.debug &
- RD_KAFKA_DBG_PROTOCOL) ? LOG_DEBUG : 0;
- size_t message_end;
-
- rd_kafka_buf_read_i64(rkbuf, &hdr.Offset);
- rd_kafka_buf_read_i32(rkbuf, &hdr.MessageSize);
- message_end = rd_slice_offset(&rkbuf->rkbuf_reader) + hdr.MessageSize;
-
- rd_kafka_buf_read_i32(rkbuf, &hdr.Crc);
- if (!rd_slice_narrow_copy_relative(&rkbuf->rkbuf_reader, &crc_slice,
- hdr.MessageSize - 4))
- rd_kafka_buf_check_len(rkbuf, hdr.MessageSize - 4);
-
- rd_kafka_buf_read_i8(rkbuf, &hdr.MagicByte);
- rd_kafka_buf_read_i8(rkbuf, &hdr.Attributes);
-
- if (hdr.MagicByte == 1) { /* MsgVersion */
- rd_kafka_buf_read_i64(rkbuf, &hdr.Timestamp);
- hdrsize += 8;
- /* MsgVersion 1 has relative offsets for compressed
MessageSets*/
- if (!(hdr.Attributes & RD_KAFKA_MSG_ATTR_COMPRESSION_MASK) &&
- msetr->msetr_relative_offsets) {
- relative_offsets = 1;
- reloff_str = "relative ";
- }
- } else
- hdr.Timestamp = 0;
-
- /* Verify MessageSize */
- if (unlikely(hdr.MessageSize < (ssize_t)hdrsize))
- rd_kafka_buf_parse_fail(rkbuf,
- "Message at %soffset %"PRId64
- " MessageSize %"PRId32
- " < hdrsize %"PRIusz,
- reloff_str,
- hdr.Offset, hdr.MessageSize, hdrsize);
-
- /* Early check for partial messages */
- rd_kafka_buf_check_len(rkbuf, hdr.MessageSize - hdrsize);
-
- if (rkb->rkb_rk->rk_conf.check_crcs) {
- /* Verify CRC32 if desired. */
- uint32_t calc_crc;
-
- calc_crc = rd_slice_crc32(&crc_slice);
- rd_dassert(rd_slice_remains(&crc_slice) == 0);
-
- if (unlikely(hdr.Crc != calc_crc)) {
- /* Propagate CRC error to application and
- * continue with next message. */
- rd_kafka_q_op_err(&msetr->msetr_rkq,
- RD_KAFKA_OP_CONSUMER_ERR,
- RD_KAFKA_RESP_ERR__BAD_MSG,
- msetr->msetr_tver->version,
- rktp,
- hdr.Offset,
- "Message at %soffset %"PRId64
- " (%"PRId32" bytes) "
- "failed CRC32 check "
- "(original 0x%"PRIx32" != "
- "calculated 0x%"PRIx32")",
- reloff_str, hdr.Offset,
- hdr.MessageSize, hdr.Crc, calc_crc);
- rd_kafka_buf_skip_to(rkbuf, message_end);
- rd_atomic64_add(&rkb->rkb_c.rx_err, 1);
- /* Continue with next message */
- return RD_KAFKA_RESP_ERR_NO_ERROR;
- }
- }
-
-
- /* Extract key */
- rd_kafka_buf_read_bytes(rkbuf, &Key);
-
- /* Extract Value */
- rd_kafka_buf_read_bytes(rkbuf, &Value);
- Value_len = RD_KAFKAP_BYTES_LEN(&Value);
-
- /* MessageSets may contain offsets earlier than we
- * requested (compressed MessageSets in particular),
- * drop the earlier messages.
- * Note: the inner offset may only be trusted for
- * absolute offsets. KIP-31 introduced
- * ApiVersion 2 that maintains relative offsets
- * of compressed messages and the base offset
- * in the outer message is the offset of
- * the *LAST* message in the MessageSet.
- * This requires us to assign offsets
- * after all messages have been read from
- * the messageset, and it also means
- * we cant perform this offset check here
- * in that case. */
- if (!relative_offsets &&
- hdr.Offset < rktp->rktp_offsets.fetch_offset)
- return RD_KAFKA_RESP_ERR_NO_ERROR; /* Continue with next msg */
-
- /* Handle compressed MessageSet */
- if (unlikely(hdr.Attributes & RD_KAFKA_MSG_ATTR_COMPRESSION_MASK))
- return rd_kafka_msgset_reader_decompress(
- msetr, hdr.MagicByte, hdr.Attributes, hdr.Timestamp,
- hdr.Offset, Value.data, Value_len);
-
-
- /* Pure uncompressed message, this is the innermost
- * handler after all compression and cascaded
- * MessageSets have been peeled off. */
-
- /* Create op/message container for message. */
- rko = rd_kafka_op_new_fetch_msg(&rkm, rktp, msetr->msetr_tver->version,
- rkbuf,
- hdr.Offset,
- (size_t)RD_KAFKAP_BYTES_LEN(&Key),
- RD_KAFKAP_BYTES_IS_NULL(&Key) ?
- NULL : Key.data,
- (size_t)RD_KAFKAP_BYTES_LEN(&Value),
- RD_KAFKAP_BYTES_IS_NULL(&Value) ?
- NULL : Value.data);
-
- /* Assign message timestamp.
- * If message was in a compressed MessageSet and the outer/wrapper
- * Message.Attribute had a LOG_APPEND_TIME set, use the
- * outer timestamp */
- if (msetr->msetr_outer.tstype == RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME) {
- rkm->rkm_timestamp = msetr->msetr_outer.timestamp;
- rkm->rkm_tstype = msetr->msetr_outer.tstype;
-
- } else if (hdr.MagicByte >= 1 && hdr.Timestamp) {
- rkm->rkm_timestamp = hdr.Timestamp;
- if (hdr.Attributes & RD_KAFKA_MSG_ATTR_LOG_APPEND_TIME)
- rkm->rkm_tstype = RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME;
- else
- rkm->rkm_tstype = RD_KAFKA_TIMESTAMP_CREATE_TIME;
- }
-
- /* Enqueue message on temporary queue */
- rd_kafka_q_enq(&msetr->msetr_rkq, rko);
- msetr->msetr_msgcnt++;
-
- return RD_KAFKA_RESP_ERR_NO_ERROR; /* Continue */
-
- err_parse:
- /* Count all parse errors as partial message errors. */
- rd_atomic64_add(&msetr->msetr_rkb->rkb_c.rx_partial, 1);
- return rkbuf->rkbuf_err;
-}
-
-
-
-
-/**
- * @brief Message parser for MsgVersion v2
- */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_msg_v2 (rd_kafka_msgset_reader_t *msetr) {
- rd_kafka_buf_t *rkbuf = msetr->msetr_rkbuf;
- rd_kafka_toppar_t *rktp = msetr->msetr_rktp;
- struct {
- int64_t Length;
- int64_t MsgAttributes; /* int8_t, but int64 req. for varint */
- int64_t TimestampDelta;
- int64_t OffsetDelta;
- int64_t Offset; /* Absolute offset */
- rd_kafkap_bytes_t Key;
- rd_kafkap_bytes_t Value;
- int64_t HeaderCnt;
- } hdr;
- rd_kafka_op_t *rko;
- rd_kafka_msg_t *rkm;
- /* Only log decoding errors if protocol debugging enabled. */
- int log_decode_errors = (rkbuf->rkbuf_rkb->rkb_rk->rk_conf.debug &
- RD_KAFKA_DBG_PROTOCOL) ? LOG_DEBUG : 0;
- size_t message_end;
-
- rd_kafka_buf_read_varint(rkbuf, &hdr.Length);
- message_end = rd_slice_offset(&rkbuf->rkbuf_reader) + hdr.Length;
- rd_kafka_buf_read_varint(rkbuf, &hdr.MsgAttributes);
-
- rd_kafka_buf_read_varint(rkbuf, &hdr.TimestampDelta);
- rd_kafka_buf_read_varint(rkbuf, &hdr.OffsetDelta);
- hdr.Offset = msetr->msetr_v2_hdr->BaseOffset + hdr.OffsetDelta;
-
- /* Skip message if outdated */
- if (hdr.Offset < rktp->rktp_offsets.fetch_offset) {
- rd_rkb_dbg(msetr->msetr_rkb, MSG, "MSG",
- "Skip offset %"PRId64" < fetch_offset %"PRId64,
- hdr.Offset, rktp->rktp_offsets.fetch_offset);
- rd_kafka_buf_skip_to(rkbuf, message_end);
- return RD_KAFKA_RESP_ERR_NO_ERROR; /* Continue with next msg */
- }
-
- rd_kafka_buf_read_bytes_varint(rkbuf, &hdr.Key);
-
- rd_kafka_buf_read_bytes_varint(rkbuf, &hdr.Value);
-
- /* Ignore headers for now */
- rd_kafka_buf_skip_to(rkbuf, message_end);
-
- /* Create op/message container for message. */
- rko = rd_kafka_op_new_fetch_msg(&rkm,
- rktp, msetr->msetr_tver->version,
rkbuf,
- hdr.Offset,
- (size_t)RD_KAFKAP_BYTES_LEN(&hdr.Key),
- RD_KAFKAP_BYTES_IS_NULL(&hdr.Key) ?
- NULL : hdr.Key.data,
-
(size_t)RD_KAFKAP_BYTES_LEN(&hdr.Value),
- RD_KAFKAP_BYTES_IS_NULL(&hdr.Value) ?
- NULL : hdr.Value.data);
-
- /* Set timestamp.
- *
- * When broker assigns the timestamps (LOG_APPEND_TIME) it will
- * assign the same timestamp for all messages in a MessageSet
- * using MaxTimestamp.
- */
- if ((msetr->msetr_v2_hdr->Attributes &
- RD_KAFKA_MSG_ATTR_LOG_APPEND_TIME) ||
- (hdr.MsgAttributes & RD_KAFKA_MSG_ATTR_LOG_APPEND_TIME)) {
- rkm->rkm_tstype = RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME;
- rkm->rkm_timestamp = msetr->msetr_v2_hdr->MaxTimestamp;
- } else {
- rkm->rkm_tstype = RD_KAFKA_TIMESTAMP_CREATE_TIME;
- rkm->rkm_timestamp =
- msetr->msetr_v2_hdr->BaseTimestamp +
hdr.TimestampDelta;
- }
-
-
- /* Enqueue message on temporary queue */
- rd_kafka_q_enq(&msetr->msetr_rkq, rko);
- msetr->msetr_msgcnt++;
-
- return RD_KAFKA_RESP_ERR_NO_ERROR;
-
- err_parse:
- /* Count all parse errors as partial message errors. */
- rd_atomic64_add(&msetr->msetr_rkb->rkb_c.rx_partial, 1);
- return rkbuf->rkbuf_err;
-}
-
-
-/**
- * @brief Read v2 messages from current buffer position.
- */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_msgs_v2 (rd_kafka_msgset_reader_t *msetr) {
- while (rd_kafka_buf_read_remain(msetr->msetr_rkbuf)) {
- rd_kafka_resp_err_t err;
- err = rd_kafka_msgset_reader_msg_v2(msetr);
- if (unlikely(err))
- return err;
- }
-
- return RD_KAFKA_RESP_ERR_NO_ERROR;
-}
-
-
-
-/**
- * @brief MessageSet reader for MsgVersion v2 (FetchRequest v4)
- */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_v2 (rd_kafka_msgset_reader_t *msetr) {
- rd_kafka_buf_t *rkbuf = msetr->msetr_rkbuf;
- rd_kafka_toppar_t *rktp = msetr->msetr_rktp;
- struct msgset_v2_hdr hdr;
- rd_slice_t save_slice;
- rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;
- size_t len_start;
- size_t payload_size;
- int64_t LastOffset; /* Last absolute Offset in MessageSet header */
- /* Only log decoding errors if protocol debugging enabled. */
- int log_decode_errors = (rkbuf->rkbuf_rkb->rkb_rk->rk_conf.debug &
- RD_KAFKA_DBG_PROTOCOL) ? LOG_DEBUG : 0;
-
- rd_kafka_buf_read_i64(rkbuf, &hdr.BaseOffset);
- rd_kafka_buf_read_i32(rkbuf, &hdr.Length);
- len_start = rd_slice_offset(&rkbuf->rkbuf_reader);
-
- if (unlikely(hdr.Length < RD_KAFKAP_MSGSET_V2_SIZE - 8 - 4))
- rd_kafka_buf_parse_fail(rkbuf,
- "%s [%"PRId32"] "
- "MessageSet at offset %"PRId64
- " length %"PRId32" < header size %d",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- hdr.BaseOffset, hdr.Length,
- RD_KAFKAP_MSGSET_V2_SIZE - 8 - 4);
-
- rd_kafka_buf_read_i32(rkbuf, &hdr.PartitionLeaderEpoch);
- rd_kafka_buf_read_i8(rkbuf, &hdr.MagicByte);
- rd_kafka_buf_read_i32(rkbuf, &hdr.Crc);
-
- if (msetr->msetr_rkb->rkb_rk->rk_conf.check_crcs) {
- /* Verify CRC32C if desired. */
- uint32_t calc_crc;
- rd_slice_t crc_slice;
- size_t crc_len = hdr.Length-4-1-4;
-
- if (!rd_slice_narrow_copy_relative(
- &rkbuf->rkbuf_reader,
- &crc_slice, crc_len))
- rd_kafka_buf_check_len(rkbuf, crc_len);
-
- calc_crc = rd_slice_crc32c(&crc_slice);
-
- if (unlikely((uint32_t)hdr.Crc != calc_crc)) {
- /* Propagate CRC error to application and
- * continue with next message. */
- rd_kafka_q_op_err(&msetr->msetr_rkq,
- RD_KAFKA_OP_CONSUMER_ERR,
- RD_KAFKA_RESP_ERR__BAD_MSG,
- msetr->msetr_tver->version,
- rktp,
- hdr.BaseOffset,
- "MessageSet at offset %"PRId64
- " (%"PRId32" bytes) "
- "failed CRC32C check "
- "(original 0x%"PRIx32" != "
- "calculated 0x%"PRIx32")",
- hdr.BaseOffset,
- hdr.Length, hdr.Crc, calc_crc);
- rd_kafka_buf_skip_to(rkbuf, crc_len);
- rd_atomic64_add(&msetr->msetr_rkb->rkb_c.rx_err, 1);
- return RD_KAFKA_RESP_ERR_NO_ERROR;
- }
- }
-
- rd_kafka_buf_read_i16(rkbuf, &hdr.Attributes);
- rd_kafka_buf_read_i32(rkbuf, &hdr.LastOffsetDelta);
- LastOffset = hdr.BaseOffset + hdr.LastOffsetDelta;
- rd_kafka_buf_read_i64(rkbuf, &hdr.BaseTimestamp);
- rd_kafka_buf_read_i64(rkbuf, &hdr.MaxTimestamp);
- rd_kafka_buf_read_i64(rkbuf, &hdr.PID);
- rd_kafka_buf_read_i16(rkbuf, &hdr.ProducerEpoch);
- rd_kafka_buf_read_i32(rkbuf, &hdr.BaseSequence);
- rd_kafka_buf_read_i32(rkbuf, &hdr.RecordCount);
-
- /* Payload size is hdr.Length - MessageSet headers */
- payload_size = hdr.Length - (rd_slice_offset(&rkbuf->rkbuf_reader) -
- len_start);
-
- if (unlikely(payload_size > rd_kafka_buf_read_remain(rkbuf)))
- rd_kafka_buf_parse_fail(rkbuf,
- "%s [%"PRId32"] "
- "MessageSet at offset %"PRId64
- " payload size %"PRIusz
- " > %"PRIusz" remaining bytes",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- hdr.BaseOffset, payload_size,
- rd_kafka_buf_read_remain(rkbuf));
-
- /* If entire MessageSet contains old outdated offsets, skip it. */
- if (LastOffset < rktp->rktp_offsets.fetch_offset) {
- rd_kafka_buf_skip(rkbuf, payload_size);
- goto done;
- }
-
- /* Ignore control messages */
- if (unlikely((hdr.Attributes & RD_KAFKA_MSGSET_V2_ATTR_CONTROL))) {
- rd_kafka_buf_skip(rkbuf, payload_size);
- goto done;
- }
-
- msetr->msetr_v2_hdr = &hdr;
-
- /* Handle compressed MessageSet */
- if (hdr.Attributes & RD_KAFKA_MSG_ATTR_COMPRESSION_MASK) {
- const void *compressed;
-
- compressed = rd_slice_ensure_contig(&rkbuf->rkbuf_reader,
- payload_size);
- rd_assert(compressed);
-
- err = rd_kafka_msgset_reader_decompress(
- msetr, 2/*MsgVersion v2*/, hdr.Attributes,
- hdr.BaseTimestamp, hdr.BaseOffset,
- compressed, payload_size);
- if (err)
- goto err;
-
- } else {
- /* Read uncompressed messages */
-
- /* Save original slice, reduce size of the current one to
- * be limited by the MessageSet.Length, and then start reading
- * messages until the lesser slice is exhausted. */
- if (!rd_slice_narrow_relative(&rkbuf->rkbuf_reader,
- &save_slice, payload_size))
- rd_kafka_buf_check_len(rkbuf, payload_size);
-
- /* Read messages */
- err = rd_kafka_msgset_reader_msgs_v2(msetr);
-
- /* Restore wider slice */
- rd_slice_widen(&rkbuf->rkbuf_reader, &save_slice);
-
- if (unlikely(err))
- goto err;
- }
-
-
- done:
- /* Set the next fetch offset to the MessageSet header's last offset + 1
- * to avoid getting stuck on compacted MessageSets where the last
- * Message in the MessageSet has an Offset < MessageSet header's
- * last offset. See KAFKA-5443 */
- if (likely(LastOffset >= msetr->msetr_rktp->rktp_offsets.fetch_offset))
- msetr->msetr_rktp->rktp_offsets.fetch_offset = LastOffset + 1;
-
- msetr->msetr_v2_hdr = NULL;
-
- return RD_KAFKA_RESP_ERR_NO_ERROR;
-
- err_parse:
- /* Count all parse errors as partial message errors. */
- rd_atomic64_add(&msetr->msetr_rkb->rkb_c.rx_partial, 1);
- err = rkbuf->rkbuf_err;
- /* FALLTHRU */
- err:
- msetr->msetr_v2_hdr = NULL;
- return err;
-}
-
-
-
-/**
- * @brief Parse and read messages from msgset reader buffer.
- */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader (rd_kafka_msgset_reader_t *msetr) {
- rd_kafka_buf_t *rkbuf = msetr->msetr_rkbuf;
- rd_kafka_toppar_t *rktp = msetr->msetr_rktp;
- rd_kafka_resp_err_t (*reader[]) (rd_kafka_msgset_reader_t *) = {
- /* Indexed by MsgVersion/MagicByte, pointing to
- * a Msg(Set)Version reader */
- [0] = rd_kafka_msgset_reader_msg_v0_1,
- [1] = rd_kafka_msgset_reader_msg_v0_1,
- [2] = rd_kafka_msgset_reader_v2
- };
- rd_kafka_resp_err_t err;
- /* Only log decoding errors if protocol debugging enabled. */
- int log_decode_errors = (rkbuf->rkbuf_rkb->rkb_rk->rk_conf.debug &
- RD_KAFKA_DBG_PROTOCOL) ? LOG_DEBUG : 0;
- int8_t MagicByte;
- size_t read_offset = rd_slice_offset(&rkbuf->rkbuf_reader);
-
- /* We dont know the MsgVersion at this point, peek where the
- * MagicByte resides both in MsgVersion v0..1 and v2 to
- * know which MessageSet reader to use. */
- rd_kafka_buf_peek_i8(rkbuf, read_offset+8+4+4, &MagicByte);
-
- if (unlikely(MagicByte < 0 || MagicByte > 2)) {
- int64_t Offset; /* For error logging */
- rd_kafka_buf_peek_i64(rkbuf, read_offset+0, &Offset);
-
- rd_rkb_dbg(msetr->msetr_rkb,
- MSG | RD_KAFKA_DBG_PROTOCOL | RD_KAFKA_DBG_FETCH,
- "MAGICBYTE",
- "%s [%"PRId32"]: "
- "Unsupported Message(Set) MagicByte %d at "
- "offset %"PRId64": skipping",
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- (int)MagicByte, Offset);
- if (Offset >= msetr->msetr_rktp->rktp_offsets.fetch_offset) {
- rd_kafka_q_op_err(
- &msetr->msetr_rkq,
- RD_KAFKA_OP_CONSUMER_ERR,
- RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED,
- msetr->msetr_tver->version, rktp, Offset,
- "Unsupported Message(Set) MagicByte %d "
- "at offset %"PRId64,
- (int)MagicByte, Offset);
- /* Skip message(set) */
- msetr->msetr_rktp->rktp_offsets.fetch_offset =
Offset+1;
- }
-
- return RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED;
- }
-
- /* Let version-specific reader parse MessageSets until the slice
- * is exhausted or an error occurs (typically a partial message). */
- do {
- err = reader[(int)MagicByte](msetr);
- } while (!err && rd_slice_remains(&rkbuf->rkbuf_reader) > 0);
-
- return err;
-
- err_parse:
- return RD_KAFKA_RESP_ERR_NO_ERROR;
-}
-
-
-
-/**
- * @brief MessageSet post-processing.
- *
- * @param last_offsetp will be set to the offset of the last message in the
set,
- * or -1 if not applicable.
- */
-static void rd_kafka_msgset_reader_postproc (rd_kafka_msgset_reader_t *msetr,
- int64_t *last_offsetp) {
- rd_kafka_op_t *rko;
-
- if (msetr->msetr_relative_offsets) {
- /* Update messages to absolute offsets
- * and purge any messages older than the current
- * fetch offset. */
- rd_kafka_q_fix_offsets(&msetr->msetr_rkq,
- msetr->msetr_rktp->rktp_offsets.
- fetch_offset,
- msetr->msetr_outer.offset -
- msetr->msetr_msgcnt + 1);
- }
-
- rko = rd_kafka_q_last(&msetr->msetr_rkq,
- RD_KAFKA_OP_FETCH,
- 0 /* no error ops */);
- if (rko)
- *last_offsetp = rko->rko_u.fetch.rkm.rkm_offset;
-}
-
-
-
-
-
-/**
- * @brief Run the MessageSet reader, read messages until buffer is
- * exhausted (or error encountered), enqueue parsed messages on
- * partition queue.
- *
- * @returns RD_KAFKA_RESP_ERR_NO_ERROR if MessageSet was successfully
- * or partially parsed. When other error codes are returned it
- * indicates a semi-permanent error (such as unsupported MsgVersion)
- * and the fetcher should back off this partition to avoid
- * busy-looping.
- */
-static rd_kafka_resp_err_t
-rd_kafka_msgset_reader_run (rd_kafka_msgset_reader_t *msetr) {
- rd_kafka_toppar_t *rktp = msetr->msetr_rktp;
- rd_kafka_resp_err_t err;
- int64_t last_offset = -1;
-
- /* Parse MessageSets and messages */
- err = rd_kafka_msgset_reader(msetr);
-
- if (unlikely(rd_kafka_q_len(&msetr->msetr_rkq) == 0)) {
- /* The message set didn't contain at least one full message
- * or no error was posted on the response queue.
- * This means the size limit perhaps was too tight,
- * increase it automatically. */
- if (rktp->rktp_fetch_msg_max_bytes < (1 << 30)) {
- rktp->rktp_fetch_msg_max_bytes *= 2;
- rd_rkb_dbg(msetr->msetr_rkb, FETCH, "CONSUME",
- "Topic %s [%"PRId32"]: Increasing "
- "max fetch bytes to %"PRId32,
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition,
- rktp->rktp_fetch_msg_max_bytes);
- } else if (!err) {
- rd_kafka_q_op_err(
- &msetr->msetr_rkq,
- RD_KAFKA_OP_CONSUMER_ERR,
- RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE,
- msetr->msetr_tver->version,
- rktp,
- rktp->rktp_offsets.fetch_offset,
- "Message at offset %"PRId64" "
- "might be too large to fetch, try increasing "
- "receive.message.max.bytes",
- rktp->rktp_offsets.fetch_offset);
- }
-
- } else {
- /* MessageSet post-processing. */
- rd_kafka_msgset_reader_postproc(msetr, &last_offset);
-
- /* Ignore parse errors if there was at least one
- * good message since it probably indicates a
- * partial response rather than an erroneous one. */
- if (err == RD_KAFKA_RESP_ERR__BAD_MSG &&
- msetr->msetr_msgcnt > 0)
- err = RD_KAFKA_RESP_ERR_NO_ERROR;
- }
-
- rd_rkb_dbg(msetr->msetr_rkb, MSG | RD_KAFKA_DBG_FETCH, "CONSUME",
- "Enqueue %i message(s) (%d ops) on %s [%"PRId32"] "
- "fetch queue (qlen %d, v%d, last_offset %"PRId64")",
- msetr->msetr_msgcnt, rd_kafka_q_len(&msetr->msetr_rkq),
- rktp->rktp_rkt->rkt_topic->str,
- rktp->rktp_partition, rd_kafka_q_len(&msetr->msetr_rkq),
- msetr->msetr_tver->version, last_offset);
-
- /* Concat all messages&errors onto the parent's queue
- * (the partition's fetch queue) */
- if (rd_kafka_q_concat(msetr->msetr_par_rkq, &msetr->msetr_rkq) != -1) {
- /* Update partition's fetch offset based on
- * last message's offest. */
- if (likely(last_offset != -1)) {
- rktp->rktp_offsets.fetch_offset = last_offset + 1;
- rd_atomic64_add(&rktp->rktp_c.msgs,
- msetr->msetr_msgcnt);
- }
- }
-
- rd_kafka_q_destroy(&msetr->msetr_rkq);
-
- /* Skip remaining part of slice so caller can continue
- * with next partition. */
- rd_slice_read(&msetr->msetr_rkbuf->rkbuf_reader, NULL,
- rd_slice_remains(&msetr->msetr_rkbuf->rkbuf_reader));
- return err;
-}
-
-
-
-/**
- * @brief Parse one MessageSet at the current buffer read position,
- * enqueueing messages, propagating errors, etc.
- * @remark The current rkbuf_reader slice must be limited to the MessageSet
size
- *
- * @returns see rd_kafka_msgset_reader_run()
- */
-rd_kafka_resp_err_t
-rd_kafka_msgset_parse (rd_kafka_buf_t *rkbuf,
- rd_kafka_buf_t *request,
- rd_kafka_toppar_t *rktp,
- const struct rd_kafka_toppar_ver *tver) {
- rd_kafka_msgset_reader_t msetr;
-
- rd_kafka_msgset_reader_init(&msetr, rkbuf, rktp, tver,
- rktp->rktp_fetchq);
-
- /* Parse and handle the message set */
- return rd_kafka_msgset_reader_run(&msetr);
-}
-
-
