Data loss consuming legacy (magic v0/v1) message-format topics over flexible Fetch

[giuseppelillo]

Description

A consumer on librdkafka v2.5.0+ does not read legacy message-format records
(magic byte 0 or 1, i.e. records stored on disk with
log.message.format.version <= 0.10.x) correctly when the broker speaks a
modern protocol that offers flexible Fetch (Fetch v12+, KIP-482). The
exact, reproducible behavior is:

• Each fetched legacy message set yields a single record with Key=NULL and
  Value=NULL. Legacy Key/Value length prefixes are 4-byte big-endian
  int32s, and any key/value smaller than 16 MiB starts with a 0x00 byte. The
  flexible-Fetch path decodes that prefix as a varint instead: it consumes only
  the leading 0x00, computes a compact-bytes length of -1, and treats the
  field as NULL. The record's real key and value bytes are never consumed,
  so the application receives an empty (NULL) key and value even though the
  record on the broker has both. The record's offset and timestamp are
  correct (those come from the fixed-width message header).
• The rest of the message set is then lost. Having consumed only 1 of the 4
  length bytes (and 0 payload bytes) for both Key and Value, the parser is
  misaligned by the unread (4 + keylen) + (4 + vallen) - 2 bytes. The next
  loop iteration reads Offset/MessageSize from the middle of the previous
  record's payload, producing a bogus (typically negative or multi-exabyte)
  MessageSize.
• That bogus length fails the message-set bounds check, and the failure mode
  is build-dependent:
  • Release builds: the batch parse aborts with
    RD_KAFKA_RESP_ERR__BAD_MSG (partial-message / buffer underflow) and every
    remaining record in the fetched batch is discarded.
  • Debug/assert builds: it trips
    assert(rd_slice_abs_offset(new_slice) <= new_slice->end) in
    rd_slice_narrow_copy() (rdbuf.c), aborting the process.
• Consumer can get stuck. If a garbage offset is enqueued before the parse
  fails, the fetch position jumps out of range; with the default
  auto.offset.reset, the consumer resets and re-fetches the same corrupt
  batch, looping indefinitely without making progress.

Net effect for the application: for a topic with N legacy records, instead of N
records with their real keys/values you get at most one NULL-keyed/NULL-valued
record per fetched batch, the remaining records dropped, and possibly a
non-terminating fetch/offset-reset loop — with no error returned to the
application in release builds (errors only appear as internal
_BAD_MSG/partial-message counters, or as protocol-debug log lines with
debug=fetch,msg,protocol).

Impact

Silent, near-total data loss on legacy-format topics: keys and values are
dropped (delivered as NULL), the majority of records per batch are discarded,
and the consumer may livelock. There is no application-visible error in a normal
(release) build, so the loss is undetectable without protocol debug logging.
magic v2 (RecordBatch) topics are unaffected, because v2 records read their
lengths via dedicated varint helpers rather than the flex-sensitive
read_kbytes.

Root cause (analysis)

Since v2.5.0 the consumer negotiates flexible Fetch (Fetch v12+) whenever the
broker advertises it. The FetchResponse is then flexible (KIP-482) and its
buffer carries the RD_KAFKA_OP_F_FLEXVER flag (copied from the request via
RD_KAFKA_BUF_FLAGS_RESP_COPY_MASK).

The Records/MessageSet payload inside a FetchResponse is encoded in the
fixed Kafka record format and is independent of the FetchResponse's flexible
framing. In particular, legacy magic v0/v1 Message Key/Value use plain
int32 length prefixes — not compact (uvarint) ones.

However the legacy message reader (rd_kafka_msgset_reader_msg_v0_1() in
src/rdkafka_msgset_reader.c) decodes Key/Value via
rd_kafka_buf_read_kbytes(), which is flex-sensitive:

#define rd_kafka_buf_read_kbytes(rkbuf, kbytes)                                \
        do {                                                                   \
                int32_t _klen;                                                 \
                if (!(rkbuf->rkbuf_flags & RD_KAFKA_OP_F_FLEXVER)) {           \
                        rd_kafka_buf_read_i32a(rkbuf, _klen);                  \
                } else {                                                       \
                        uint64_t _uva;                                         \
                        rd_kafka_buf_read_uvarint(rkbuf, &_uva);               \
                        _klen = ((int32_t)_uva) - 1;                           \
                }                                                              \
                ...

When FLEXVER is set it reads a compact (uvarint) length instead of the int32
the legacy format actually uses, misaligning the parse as described above.
(The fixed-width header fields — Offset, MessageSize, Crc, MagicByte,
Attributes, Timestamp — are read with non-flex readers and decode
correctly, which is why the first record's offset/timestamp survive while its
key/value are lost and subsequent records desync.)

How to reproduce

1. Start an old-enough broker. The bug needs a broker in a narrow window:
old enough to still store the legacy on-disk format via message.format.version
(pre-3.0), yet new enough to speak flexible Fetch (v12+, Kafka 2.4+). Apache
Kafka 2.8 fits — here via Confluent Platform 6.2 (Kafka 2.8.x), ZooKeeper
mode. Don't use 3.0+ images (message.format.version is a no-op there, removed
in 4.0), nor very old ones (e.g. Kafka 0.10/0.11 store v1 but lack flexible
Fetch) — neither triggers the bug.

# docker-compose.yml
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:6.2.2
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:6.2.2      # Apache Kafka 2.8.x
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

2. Create and populate a legacy-format topic. The per-topic
message.format.version=0.10.0 config forces magic v1 on disk for just this
topic. The Kafka CLI lives inside the broker container, so no local Kafka
install is required.

docker compose up -d

# Legacy (magic v1) topic:
docker compose exec kafka kafka-topics --bootstrap-server localhost:9092 \
  --create --topic legacy-v1-repro --partitions 1 --replication-factor 1 \
  --config message.format.version=0.10.0

# Produce 100 keyed records -> stored on disk as magic v1:
docker compose exec -T kafka bash -c \
  'for i in $(seq 1 100); do echo "key-$i:value-$i"; done | \
   kafka-console-producer --bootstrap-server localhost:9092 \
     --topic legacy-v1-repro --property parse.key=true --property key.separator=:'

3. Consume with stock librdkafka (>= v2.5.0). A default consumer enables
flexible Fetch automatically.

/* repro.c -- silent data loss reading a legacy (magic v0/v1) topic
 *            with librdkafka >= 2.5.0 over flexible Fetch.
 *
 *   build: cc repro.c -o repro -lrdkafka
 *   run:   ./repro localhost:9092 legacy-v1-repro 100
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <librdkafka/rdkafka.h>

static void set(rd_kafka_conf_t *c, const char *k, const char *v) {
        char e[512];
        if (rd_kafka_conf_set(c, k, v, e, sizeof(e)) != RD_KAFKA_CONF_OK) {
                fprintf(stderr, "conf %s=%s: %s\n", k, v, e);
                exit(1);
        }
}

int main(int argc, char **argv) {
        const char *brokers = argc > 1 ? argv[1] : "localhost:9092";
        const char *topic   = argc > 2 ? argv[2] : "legacy-v1-repro";
        int         produced = argc > 3 ? atoi(argv[3]) : 100;
        char errstr[512];

        rd_kafka_conf_t *conf = rd_kafka_conf_new();
        set(conf, "bootstrap.servers", brokers);
        set(conf, "group.id", "legacy-repro");
        set(conf, "auto.offset.reset", "earliest");
        set(conf, "enable.auto.commit", "false");
        set(conf, "enable.partition.eof", "true");
        /* Uncomment to watch the misaligned parse / _BAD_MSG errors:
           set(conf, "debug", "fetch,msg,protocol"); */

        rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf,
                                      errstr, sizeof(errstr));
        if (!rk) {
                fprintf(stderr, "%s\n", errstr);
                return 1;
        }
        rd_kafka_poll_set_consumer(rk);

        rd_kafka_topic_partition_list_t *sub =
            rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(sub, topic, RD_KAFKA_PARTITION_UA);
        rd_kafka_subscribe(rk, sub);
        rd_kafka_topic_partition_list_destroy(sub);

        int    delivered = 0, null_kv = 0, errors = 0;
        time_t deadline = time(NULL) + 10; /* bound: the bug may loop forever */

        while (time(NULL) < deadline) {
                rd_kafka_message_t *m = rd_kafka_consumer_poll(rk, 500);
                if (!m)
                        continue;
                if (m->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
                        rd_kafka_message_destroy(m);
                        break; /* reached end of partition */
                }
                if (m->err) {
                        errors++;
                        fprintf(stderr, "error @%lld: %s\n",
                                (long long)m->offset,
                                rd_kafka_message_errstr(m));
                        rd_kafka_message_destroy(m);
                        continue;
                }
                delivered++;
                if (m->key_len == 0 && m->len == 0)
                        null_kv++;
                printf("offset=%lld key_len=%zu value_len=%zu\n",
                       (long long)m->offset, m->key_len, m->len);
                rd_kafka_message_destroy(m);
        }

        printf("\nProduced ~%d records; DELIVERED %d (errors=%d), of which %d "
               "had NULL/empty key AND value.\n",
               produced, delivered, errors, null_kv);
        if (delivered < produced || null_kv > 0 || errors > 0)
                printf("BUG: real keys/values dropped / records lost "
                       "(delivered=%d null_kv=%d errors=%d).\n",
                       delivered, null_kv, errors);
        else
                printf("OK: all %d records delivered intact.\n", delivered);

        rd_kafka_consumer_close(rk);
        rd_kafka_destroy(rk);
        return 0;
}

Actual (buggy) output — captured against the broker above, using a
librdkafka v2.14.2 build (Apache Kafka 2.8 / cp-kafka:6.2.2). 100 records
were produced; every record that comes back has an empty key/value, interleaved
with bogus Unsupported MagicByte errors at impossible offsets, and only a
fraction are returned within the 10s window:

error @1525458452427008: Unsupported Message(Set) MagicByte 45 at offset 1525458452427008
error @1525458452427264: Unsupported Message(Set) MagicByte 45 at offset 1525458452427264
... (more such errors) ...
error @1806933429137712: Unsupported Message(Set) MagicByte 101 at offset 1806933429137712
offset=0 key_len=0 value_len=0
offset=1 key_len=0 value_len=0
offset=2 key_len=0 value_len=0
... (offsets 3..10, all key_len=0 value_len=0) ...

Produced ~100 records; DELIVERED 11 (errors=11), of which 11 had NULL/empty key AND value.
BUG: real keys/values dropped / records lost (delivered=11 null_kv=11 errors=11).

The exact counts vary with byte alignment and fetch batching, but the invariant
holds: real keys/values are dropped and most records never arrive.

Checklist

• librdkafka version (release number or git tag): affected v2.5.0 →
  v2.14.2 (current); reproduced on v2.14.2
• Apache Kafka version: any broker offering flexible Fetch (v12+) while
  serving legacy magic v0/v1 records
• librdkafka client configuration: defaults are sufficient
  (api.version.request=true, which enables flexible Fetch); e.g.
  auto.offset.reset=earliest
• Operating system: Linux x86_64, macOS
• Provide logs (with debug=fetch,msg,protocol) from librdkafka
• Provide broker log excerpts
• Critical issue (silent data loss)