diff --git a/src/rdkafka_buf.h b/src/rdkafka_buf.h
index 9682d099a0..447699bbe1 100644
--- a/src/rdkafka_buf.h
+++ b/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.
diff --git a/src/rdkafka_mock.c b/src/rdkafka_mock.c
index fee21ce1fb..cc40a8d9d0 100644
--- a/src/rdkafka_mock.c
+++ b/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;
diff --git a/src/rdkafka_msgset_reader.c b/src/rdkafka_msgset_reader.c
index 787b3c20ab..1a50309396 100644
--- a/src/rdkafka_msgset_reader.c
+++ b/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
diff --git a/src/rdkafka_proto.h b/src/rdkafka_proto.h
index 0ad21e06fc..32c4eded9e 100644
--- a/src/rdkafka_proto.h
+++ b/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.
diff --git a/tests/0187-legacy_msgset_fetch_mock.c b/tests/0187-legacy_msgset_fetch_mock.c
new file mode 100644
index 0000000000..7ea5b4dedc
--- /dev/null
+++ b/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;
+}
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 0f6a0650fe..5fe3e32134 100644
--- a/tests/CMakeLists.txt
+++ b/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
diff --git a/tests/test.c b/tests/test.c
index 664376afda..24947655db 100644
--- a/tests/test.c
+++ b/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)),
diff --git a/win32/tests/tests.vcxproj b/win32/tests/tests.vcxproj
index ddff9d7ebc..1e8f7dfae8 100644
--- a/win32/tests/tests.vcxproj
+++ b/win32/tests/tests.vcxproj
@@ -258,6 +258,7 @@
+