[Producer] what will fail the assert in rd_kafka_msgset_writer_write_msg() ?

[xlichao]

Hi everyone,

We are using rsyslog with librdkafka in our log system, but currently we suffer from unexpectly crash of the rsyslog process and the backtrace shows that the reason it quit is signal 6, Aborted.

The full backtrace indicates a assert in rd_kafka_msgset_writer_write_msg stop the whole process:

Core was generated by `/usr/local/sbin/rsyslogd -n -f /etc/rsyslog.d/rsyslog-classify.conf'.
Program terminated with signal 6, Aborted.
#0 0x00007fd5b8e4f387 in raise () from /lib64/libc.so.6

#0  0x00007fd5b8e4f387 in raise () from /lib64/libc.so.6
No symbol table info available.
#1  0x00007fd5b8e50a78 in abort () from /lib64/libc.so.6
No symbol table info available.
#2  0x00007fd5b8e481a6 in __assert_fail_base () from /lib64/libc.so.6
No symbol table info available.
#3  0x00007fd5b8e48252 in __assert_fail () from /lib64/libc.so.6
No symbol table info available.
#4  0x00007fd5b61ee057 in rd_kafka_msgset_writer_write_msg (msetw=msetw@entry=0x7fd228fc84e0, rkm=rkm@entry=0x7fd3a8004ad0, Offset=Offset@entry=0,
    MsgAttributes=MsgAttributes@entry=0 '\000', free_cb=free_cb@entry=0x0) at rdkafka_msgset_writer.c:819
        outlen = <optimized out>
        writer = {0x7fd5b61ee0d0 <rd_kafka_msgset_writer_write_msg_v0_1>, 0x7fd5b61ee0d0 <rd_kafka_msgset_writer_write_msg_v0_1>,
          0x7fd5b61ee5a0 <rd_kafka_msgset_writer_write_msg_v2>}
        actual_written = <optimized out>
        __PRETTY_FUNCTION__ = "rd_kafka_msgset_writer_write_msg"
#5  0x00007fd5b61ef8f1 in rd_kafka_msgset_writer_write_msgq (rkmq=0x7fd3a400cfa8, msetw=0x7fd228fc84e0) at rdkafka_msgset_writer.c:926
        rktp = 0x7fd3a400cec0
        max_msg_size = 50000
        int_latency_base = 56438284579063
        rkm = 0x7fd3a8004ad0
        msgcnt = 0
        rkb = 0x7fd3a80166e0
        MaxTimestamp = 1647094272594
        len = 131
#6  rd_kafka_msgset_create_ProduceRequest (rkb=rkb@entry=0x7fd3a80166e0, rktp=rktp@entry=0x7fd3a400cec0, rkmq=rkmq@entry=0x7fd3a400cfa8, pid=...,
    MessageSetSizep=MessageSetSizep@entry=0x7fd228fc86c8) at rdkafka_msgset_writer.c:1466
        msetw = {msetw_rkbuf = 0x7fd3bc007f90, msetw_ApiVersion = 6, msetw_MsgVersion = 2, msetw_features = 512, msetw_compression = RD_KAFKA_COMPRESSION_SNAPPY,
          msetw_msgcntmax = 20, msetw_messages_len = 0, msetw_messages_kvlen = 231, msetw_MessageSetSize = 61, msetw_of_MessageSetSize = 66, msetw_of_start = 70,
          msetw_relative_offsets = 1, msetw_Attributes = 0, msetw_MaxTimestamp = 0, msetw_of_CRC = 87, msetw_batch = 0x7fd3bc0080f8, msetw_firstmsg = {of = 131,
            timestamp = 1647094272594}, msetw_pid = {id = 7734661, epoch = 0}, msetw_rkb = 0x7fd3a80166e0, msetw_rktp = 0x7fd3a400cec0, msetw_msgq = 0x7fd3a400cfa8}
#7  0x00007fd5b61a5899 in rd_kafka_ProduceRequest (rkb=rkb@entry=0x7fd3a80166e0, rktp=rktp@entry=0x7fd3a400cec0, pid=...) at rdkafka_request.c:3457
        rkbuf = <optimized out>
        rkt = 0x7fd3a801b0f0
        MessageSetSize = 0
        now = 1486
        first_msg_timeout = <optimized out>
#8  0x00007fd5b615fbff in rd_kafka_toppar_producer_serve (rkb=rkb@entry=0x7fd3a80166e0, rktp=rktp@entry=0x7fd3a400cec0, pid=..., now=now@entry=56437984579050,
    next_wakeup=next_wakeup@entry=0x7fd228fc8940, do_timeout_scan=do_timeout_scan@entry=0 '\000', may_send=may_send@entry=1 '\001') at rdkafka_broker.c:3848
        cnt = 0
        r = <optimized out>
        rkm = <optimized out>
        move_cnt = <optimized out>
        max_requests = 1
        reqcnt = 0
        inflight = <optimized out>
#9  0x00007fd5b616c3de in rd_kafka_broker_produce_toppars (do_timeout_scan=<optimized out>, next_wakeup=0x7fd228fc8938, now=<optimized out>, rkb=0x7fd3a80166e0)

I checked out the commit (74b1790 v1.6.0-RC2 ) which we used to build rsyslog and found the source code of rd_kafka_msgset_writer_write_msg is:

/**
 * @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);     // <--- we failed here

        return outlen;

}

The code shows that we failed in rd_assert(outlen == actual_written) . However there are serval version of rd_kafka_msgset_writer_write_msg involved here and looks diffcult to me.

So I wonder what situation will cause the outlen != actual_written, is there someting wrong with my kafka brokers (v2.0.0)?

Hope for any reply, thanks ,

that happened quite frequence online (3~5 coredump per hour)

[edenhill]

Can you provide your producer config?

[xlichao]

Hello edenhill,
thanks for your reply. My config is :

"enable.idempotence=true",
"batch.num.messages=10000",
"message.max.bytes=50000",
"queue.buffering.max.ms=20",
"queue.buffering.max.kbytes=131072",
"compression.codec=snappy",
 "queue.buffering.max.messages=100000",
"sasl.mechanism=PLAIN",
"security.protocol=SASL_PLAINTEXT",
"sasl.username={{username}}",
"sasl.password={{password}}"

and partitions.auto also set to "on"

[edenhill]

Would it be possible for you to create a small reproducible program?

[xlichao]

We are working on it because today it happened again when one of our kafka broker servers offline due to disk faliure. We have noticed that some operation to kafka broker will cause the assert fail for a certain period and the rsyslog will crash for serval time. But we still trying to find out which operation can repeat the crash and why some faliure of partition in one or more brokers can affect the function "Write message to messageset buffer" in librdkafka.
Thank you for your attention ~

[xlichao]

Final result update after a long long time for someone maybe interested:

In omkafka plugin will keep those messages that failed to call rd_kafka_producev in a linkedlist and try to reproduce it in next time. That work fine before we add a header to the message in order to mark where the message from.

In retry period , we put a failed message with header into rd_kafka_producev and destory it when rd_kafka_producev return RD_KAFKA_RESP_ERR_NO_ERROR . Here is the problem: there is an queue inside librdkafka and this message will not be sent immediately especially when the system under hevay load. The call of rd_kafka_headers_destroy outside librdkafka will destory the struct of header and if its memory area has been reuse or override before actually sent to brokers, the size of header will become crazy. That is the reason of outlen != actual_written and core dumped.

The miscall of rd_kafka_headers_destroy will also lead to serval situation:

1. warning of double free
2. the linkedlist of message or header inside librdkafka suffer from zero pointer
3. failed with malloc()

The reproducible enviroment is simple: try to write message into a topic with single partition in 100k/s. After we fix the unnecessary call of rd_kafka_headers_destroy, that will only result in a full queue in both rsyslog and librdkafka. And the online system has been running smoothly for half a year handling 17+ million messages per second.

librdkafka is great, thanks~