Fix NULL key/value when consuming legacy MessageSets with flexible Fetch versions (v12+)

src/rdkafka_buf.h

@@ -769,6 +769,27 @@ struct rd_kafka_buf_s { /* rd_kafka_buf_t */
                         rd_kafka_buf_check_len(rkbuf, _klen);                  \
         } while (0)
 
+/**
+ * Read standard BYTES representation (4+N), regardless of the buffer's
+ * flexible-version flag: for structures that always use the non-compact
+ * encoding, such as legacy v0..v1 MessageSets in FetchResponses >= v12.
+ * The 'kbytes' will be updated to point to rkbuf data.
+ */
+#define rd_kafka_buf_read_kbytes_fixed(rkbuf, kbytes)                          \
+        do {                                                                   \
+                int32_t _klen;                                                 \
+                rd_kafka_buf_read_i32a(rkbuf, _klen);                          \
+                (kbytes)->len = _klen;                                         \
+                if (RD_KAFKAP_BYTES_IS_NULL(kbytes)) {                         \
+                        (kbytes)->data = NULL;                                 \
+                        (kbytes)->len  = 0;                                    \
+                } else if (RD_KAFKAP_BYTES_LEN(kbytes) == 0)                   \
+                        (kbytes)->data = "";                                   \
+                else if (!((kbytes)->data = rd_slice_ensure_contig(            \
+                               &(rkbuf)->rkbuf_reader, _klen)))                \
+                        rd_kafka_buf_check_len(rkbuf, _klen);                  \
+        } while (0)
+
 /**
  * @brief Read \p size bytes from buffer, setting \p *ptr to the start
  *        of the memory region.

src/rdkafka_mock.c

@@ -96,11 +96,16 @@ static void rd_kafka_mock_msgset_destroy(rd_kafka_mock_partition_t *mpart,
 /**
  * @brief Create a new msgset object with a copy of \p bytes
  *        and appends it to the partition log.
+ *
+ * @param MagicByte the MsgVersion of \p bytes: v2 MessageSets get their
+ *        BaseOffset and PartitionLeaderEpoch updated, while legacy v0..v1
+ *        MessageSets get an absolute offset assigned to each message.
  */
 static rd_kafka_mock_msgset_t *
 rd_kafka_mock_msgset_new(rd_kafka_mock_partition_t *mpart,
                          const rd_kafkap_bytes_t *bytes,
-                         size_t msgcnt) {
+                         size_t msgcnt,
+                         int8_t MagicByte) {
         rd_kafka_mock_msgset_t *mset;
         size_t totsize = sizeof(*mset) + RD_KAFKAP_BYTES_LEN(bytes);
         int64_t BaseOffset;
@@ -127,15 +132,39 @@ rd_kafka_mock_msgset_new(rd_kafka_mock_partition_t *mpart,
         memcpy((void *)mset->bytes.data, bytes->data, mset->bytes.len);
         mpart->size += mset->bytes.len;
 
-        /* Update the base Offset in the MessageSet with the
-         * actual absolute log offset. */
-        BaseOffset = htobe64(mset->first_offset);
-        memcpy((void *)mset->bytes.data, &BaseOffset, sizeof(BaseOffset));
-        /* Update the base PartitionLeaderEpoch in the MessageSet with the
-         * actual partition leader epoch. */
-        PartitionLeaderEpoch = htobe32(mset->leader_epoch);
-        memcpy(((char *)mset->bytes.data) + 12, &PartitionLeaderEpoch,
-               sizeof(PartitionLeaderEpoch));
+        if (MagicByte >= 2) {
+                /* Update the base Offset in the MessageSet with the
+                 * actual absolute log offset. */
+                BaseOffset = htobe64(mset->first_offset);
+                memcpy((void *)mset->bytes.data, &BaseOffset,
+                       sizeof(BaseOffset));
+                /* Update the base PartitionLeaderEpoch in the MessageSet with
+                 * the actual partition leader epoch. */
+                PartitionLeaderEpoch = htobe32(mset->leader_epoch);
+                memcpy(((char *)mset->bytes.data) + 12, &PartitionLeaderEpoch,
+                       sizeof(PartitionLeaderEpoch));
+        } else {
+                /* Legacy v0..v1 MessageSet: update the Offset of each
+                 * message in the set with the actual absolute log offset.
+                 * The MessageSet layout was validated by the caller. */
+                size_t of      = 0;
+                int64_t offset = mset->first_offset;
+                char *data     = (char *)mset->bytes.data;
+
+                while (of + RD_KAFKAP_MESSAGESET_V0_HDR_SIZE <=
+                       (size_t)mset->bytes.len) {
+                        int64_t Offset = htobe64(offset++);
+                        int32_t MessageSize;
+
+                        memcpy(data + of, &Offset, sizeof(Offset));
+                        memcpy(&MessageSize,
+                               data + of + RD_KAFKAP_MSGSET_V0_OF_MessageSize,
+                               sizeof(MessageSize));
+                        MessageSize = be32toh(MessageSize);
+                        of += RD_KAFKAP_MESSAGESET_V0_HDR_SIZE +
+                              (size_t)MessageSize;
+                }
+        }
 
         /* Remove old msgsets until within limits */
         while (mpart->cnt > 1 &&
@@ -423,7 +452,7 @@ void rd_kafka_mock_partition_write_control_batch(
         bytes.data = buf;
 
         /* msgset_new copies the bytes, assigns offsets, and appends. */
-        rd_kafka_mock_msgset_new(mpart, &bytes, 1);
+        rd_kafka_mock_msgset_new(mpart, &bytes, 1, 2 /*MagicByte*/);
 
         rd_kafka_dbg(mpart->topic->cluster->rk, MOCK, "MOCK",
                      "Wrote %s control batch to %s [%" PRId32
@@ -551,6 +580,58 @@ rd_kafka_mock_validate_records(rd_kafka_mock_partition_t *mpart,
         return rkbuf->rkbuf_err;
 }
 
+/**
+ * @brief Parse and validate an uncompressed legacy (MsgVersion v0..v1)
+ *        MessageSet and return the number of messages it contains
+ *        in \p msgcntp.
+ *
+ * Compressed legacy MessageSets are not supported and are rejected
+ * with RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION.
+ */
+static rd_kafka_resp_err_t
+rd_kafka_mock_msgset_legacy_count(rd_kafka_buf_t *rkbuf,
+                                  size_t len,
+                                  size_t *msgcntp) {
+        const int log_decode_errors = LOG_ERR;
+        size_t of                   = 0;
+        size_t msgcnt               = 0;
+
+        while (of < len) {
+                int32_t MessageSize;
+                int8_t Attributes;
+
+                if (of + RD_KAFKAP_MESSAGE_V0_OVERHEAD > len)
+                        return RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE;
+
+                rd_kafka_buf_peek_i32(rkbuf,
+                                      of + RD_KAFKAP_MSGSET_V0_OF_MessageSize,
+                                      &MessageSize);
+                rd_kafka_buf_peek_i8(
+                    rkbuf, of + RD_KAFKAP_MSGSET_V0_OF_Attributes, &Attributes);
+
+                if (Attributes & RD_KAFKA_MSG_ATTR_COMPRESSION_MASK)
+                        return RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION;
+
+                if (MessageSize < RD_KAFKAP_MESSAGE_V0_HDR_SIZE ||
+                    of + RD_KAFKAP_MESSAGESET_V0_HDR_SIZE +
+                            (size_t)MessageSize >
+                        len)
+                        return RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE;
+
+                of += RD_KAFKAP_MESSAGESET_V0_HDR_SIZE + (size_t)MessageSize;
+                msgcnt++;
+        }
+
+        if (msgcnt == 0)
+                return RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE;
+
+        *msgcntp = msgcnt;
+        return RD_KAFKA_RESP_ERR_NO_ERROR;
+
+err_parse:
+        return rkbuf->rkbuf_err;
+}
+
 /**
  * @brief Append the MessageSets in \p bytes to the \p mpart partition log.
  *
@@ -577,10 +658,29 @@ rd_kafka_mock_partition_log_append(rd_kafka_mock_partition_t *mpart,
 
         rd_kafka_buf_peek_i8(rkbuf, RD_KAFKAP_MSGSET_V2_OF_MagicByte,
                              &MagicByte);
-        if (MagicByte != 2) {
-                /* We only support MsgVersion 2 for now */
-                err = RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION;
-                goto err;
+        if (MagicByte < 2) {
+                /* Legacy uncompressed v0..v1 MessageSet, as produced to
+                 * clusters with log.message.format.version < 0.11. */
+                size_t legacy_msgcnt;
+
+                err = rd_kafka_mock_msgset_legacy_count(
+                    rkbuf, RD_KAFKAP_BYTES_LEN(records), &legacy_msgcnt);
+                if (err)
+                        goto err;
+
+                rd_kafka_buf_destroy(rkbuf);
+
+                mset = rd_kafka_mock_msgset_new(mpart, records, legacy_msgcnt,
+                                                MagicByte);
+
+                *BaseOffset = mset->first_offset;
+
+                mtx_lock(&mpart->topic->cluster->lock);
+                rd_kafka_mock_partition_update_lso(mpart,
+                                                   mpart->topic->cluster);
+                mtx_unlock(&mpart->topic->cluster->lock);
+
+                return RD_KAFKA_RESP_ERR_NO_ERROR;
         }
 
         rd_kafka_buf_peek_i32(rkbuf, RD_KAFKAP_MSGSET_V2_OF_RecordCount,
@@ -606,7 +706,8 @@ rd_kafka_mock_partition_log_append(rd_kafka_mock_partition_t *mpart,
 
         rd_kafka_buf_destroy(rkbuf);
 
-        mset = rd_kafka_mock_msgset_new(mpart, records, (size_t)RecordCount);
+        mset = rd_kafka_mock_msgset_new(mpart, records, (size_t)RecordCount,
+                                        MagicByte);
 
         *BaseOffset = mset->first_offset;
 

src/rdkafka_msgset_reader.c

@@ -684,11 +684,15 @@ rd_kafka_msgset_reader_msg_v0_1(rd_kafka_msgset_reader_t *msetr) {
         }
 
 
+        /* Key and Value always use the fixed-width BYTES encoding, also
+         * when this MessageSet is contained in a flexible-version
+         * FetchResponse (v12+) whose buffer has the flexver flag set. */
+
         /* Extract key */
-        rd_kafka_buf_read_kbytes(rkbuf, &Key);
+        rd_kafka_buf_read_kbytes_fixed(rkbuf, &Key);
 
         /* Extract Value */
-        rd_kafka_buf_read_kbytes(rkbuf, &Value);
+        rd_kafka_buf_read_kbytes_fixed(rkbuf, &Value);
         Value_len = RD_KAFKAP_BYTES_LEN(&Value);
 
         /* MessageSets may contain offsets earlier than we

src/rdkafka_proto.h

@@ -578,6 +578,11 @@ typedef struct rd_kafka_buf_s rd_kafka_buf_t;
 #define RD_KAFKAP_MSGSET_V2_OF_RecordCount                                     \
         (8 + 4 + 4 + 1 + 4 + 2 + 4 + 8 + 8 + 8 + 2 + 4)
 
+/* Byte offsets for legacy MsgVersion v0..v1 MessageSet fields */
+#define RD_KAFKAP_MSGSET_V0_OF_MessageSize (8)
+#define RD_KAFKAP_MSGSET_V0_OF_MagicByte   (8 + 4 + 4)
+#define RD_KAFKAP_MSGSET_V0_OF_Attributes  (8 + 4 + 4 + 1)
+
 
 /**
  * @struct Struct representing UUID protocol primitive type.

tests/0187-legacy_msgset_fetch_mock.c

@@ -0,0 +1,114 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2026, Confluent Inc.
+ * 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 "test.h"
+
+/**
+ * @name Consuming legacy (MsgVersion v0..v1) MessageSets with flexible
+ *       Fetch versions (v12+), using the mock broker.
+ *
+ * Brokers with log.message.format.version < 0.11 (e.g. during a rolling
+ * upgrade from an old cluster) store and return MessageSets in the legacy
+ * v0/v1 format, regardless of the Fetch version used by the consumer.
+ * The legacy MessageSet payload always uses non-compact (fixed-width)
+ * encodings, even when the enclosing FetchResponse is a flexible version.
+ *
+ * This is a regression test: messages were returned to the application
+ * with NULL key and value, because the Key and Value fields were parsed
+ * as compact bytes (varint length) when the FetchResponse was a flexible
+ * version.
+ */
+
+
+/**
+ * @brief Produce and consume with the producer's ProduceRequest capped at
+ *        \p max_produce_version, so that it writes legacy MessageSets:
+ *        v2 => MsgVersion 1, v1 => MsgVersion 0.
+ */
+static void do_test_legacy_msgset_fetch(int16_t max_produce_version) {
+        const char *topic = "legacy_msgset";
+        const int msgcnt  = 100;
+        rd_kafka_mock_cluster_t *mcluster;
+        const char *bootstraps;
+        rd_kafka_conf_t *conf;
+        rd_kafka_t *c;
+        uint64_t testid = test_id_generate();
+        test_msgver_t mv;
+
+        SUB_TEST_QUICK("max ProduceRequest v%" PRId16, max_produce_version);
+
+        mcluster = test_mock_cluster_new(1, &bootstraps);
+
+        rd_kafka_mock_topic_create(mcluster, topic, 1, 1);
+
+        /* Cap the ProduceRequest version so the producer selects a legacy
+         * MsgVersion (the MSGVER1 feature requires Produce >= v2 and
+         * MSGVER2 requires Produce >= v3), while Fetch stays at a
+         * flexible version (v12+). */
+        TEST_CALL_ERR__(rd_kafka_mock_set_apiversion(mcluster, 0 /*Produce*/, 0,
+                                                     max_produce_version));
+
+        /* Seed the topic with messages */
+        test_produce_msgs_easy_v(topic, testid, 0, 0, msgcnt, 100,
+                                 "bootstrap.servers", bootstraps, NULL);
+
+        test_conf_init(&conf, NULL, 30);
+        test_conf_set(conf, "bootstrap.servers", bootstraps);
+        test_conf_set(conf, "auto.offset.reset", "earliest");
+
+        c = test_create_consumer("legacy_msgset_group", NULL, conf, NULL);
+
+        test_consumer_subscribe(c, topic);
+
+        /* Verify that all messages are consumed with intact payloads:
+         * with the bug the messages arrive with NULL key and value. */
+        test_msgver_init(&mv, testid);
+        test_consumer_poll("consume.legacy", c, testid, -1, 0, msgcnt, &mv);
+        test_msgver_verify("verify.legacy", &mv, TEST_MSGVER_ALL, 0, msgcnt);
+        test_msgver_clear(&mv);
+
+        test_consumer_close(c);
+        rd_kafka_destroy(c);
+
+        test_mock_cluster_destroy(mcluster);
+
+        SUB_TEST_PASS();
+}
+
+
+int main_0187_legacy_msgset_fetch_mock(int argc, char **argv) {
+        TEST_SKIP_MOCK_CLUSTER(0);
+
+        /* MsgVersion v1 */
+        do_test_legacy_msgset_fetch(2);
+
+        /* MsgVersion v0 */
+        do_test_legacy_msgset_fetch(1);
+
+        return 0;
+}

tests/CMakeLists.txt

@@ -165,6 +165,7 @@ set(
     0184-share_consumer_topic_recreate.c
     0185-share_consumer_max_poll_interval.c
     0186-share_consumer_fatal_error.c
+    0187-legacy_msgset_fetch_mock.c
     0190-share_consumer_telemetry.c
     8000-idle.cpp
     8001-fetch_from_follower_mock_manual.c

tests/test.c

@@ -313,6 +313,7 @@ _TEST_DECL(0184_share_consumer_topic_recreate);
 _TEST_DECL(0184_share_consumer_topic_recreate_local);
 _TEST_DECL(0185_share_consumer_max_poll_interval);
 _TEST_DECL(0186_share_consumer_fatal_error);
+_TEST_DECL(0187_legacy_msgset_fetch_mock);
 _TEST_DECL(0190_share_consumer_telemetry);
 
 /* Manual tests */
@@ -609,6 +610,7 @@ struct test tests[] = {
     _TEST(0184_share_consumer_topic_recreate_local, TEST_F_LOCAL),
     _TEST(0185_share_consumer_max_poll_interval, 0, TEST_BRKVER(4, 2, 0, 0)),
     _TEST(0186_share_consumer_fatal_error, TEST_F_LOCAL),
+    _TEST(0187_legacy_msgset_fetch_mock, TEST_F_LOCAL),
     _TEST(0190_share_consumer_telemetry,
           TEST_F_MANUAL,
           TEST_BRKVER(4, 2, 0, 0)),

win32/tests/tests.vcxproj

@@ -258,6 +258,7 @@
     <ClCompile Include="..\..\tests\0184-share_consumer_topic_recreate.c" />
     <ClCompile Include="..\..\tests\0185-share_consumer_max_poll_interval.c" />
     <ClCompile Include="..\..\tests\0186-share_consumer_fatal_error.c" />
+    <ClCompile Include="..\..\tests\0187-legacy_msgset_fetch_mock.c" />
     <ClCompile Include="..\..\tests\0190-share_consumer_telemetry.c" />
     <ClCompile Include="..\..\tests\8000-idle.cpp" />
     <ClCompile Include="..\..\tests\8001-fetch_from_follower_mock_manual.c" />