This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 2894368e1 feat(gateways): map Kafka records to Iggy messages and back
(#4229)
2894368e1 is described below
commit 2894368e1bbf0452c18e1a5bee1478732b6a6f7c
Author: Krishna Vishal <[email protected]>
AuthorDate: Tue Sep 22 17:25:31 2026 +0530
feat(gateways): map Kafka records to Iggy messages and back (#4229)
---
Cargo.lock | 29 +
Cargo.toml | 5 +
gateways/kafka/Cargo.toml | 16 +-
gateways/kafka/docs/BRIDGE_MAPPING.md | 195 +++-
gateways/kafka/src/lib.rs | 1 +
gateways/kafka/src/records.rs | 1830 +++++++++++++++++++++++++++++++++
6 files changed, 2050 insertions(+), 26 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index c0a533ae0..36043f52b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7017,13 +7017,18 @@ dependencies = [
name = "iggy-gateway-kafka"
version = "0.1.0"
dependencies = [
+ "anyhow",
"bytes",
+ "crc32c",
+ "flate2",
"iggy",
"kafka-protocol",
"libc",
+ "lz4",
"secrecy",
"serde",
"serial_test",
+ "snap",
"socket2 0.6.5",
"tempfile",
"thiserror 2.0.20",
@@ -7033,6 +7038,7 @@ dependencies = [
"tracing",
"tracing-appender",
"tracing-subscriber",
+ "zstd 0.13.3",
]
[[package]]
@@ -8168,8 +8174,12 @@ dependencies = [
"bytes",
"crc",
"crc32c",
+ "flate2",
"indexmap 2.14.2",
+ "lz4",
+ "snap",
"uuid",
+ "zstd 0.13.3",
]
[[package]]
@@ -8649,6 +8659,25 @@ version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f"
+[[package]]
+name = "lz4"
+version = "1.28.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a20b523e860d03443e98350ceaac5e71c6ba89aea7d960769ec3ce37f4de5af4"
+dependencies = [
+ "lz4-sys",
+]
+
+[[package]]
+name = "lz4-sys"
+version = "1.11.1+lz4-1.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "6bd8c0d6c6ed0cd30b3652886bb8711dc4bb01d637a68105a3d5158039b418e6"
+dependencies = [
+ "cc",
+ "libc",
+]
+
[[package]]
name = "lz4_flex"
version = "0.13.1"
diff --git a/Cargo.toml b/Cargo.toml
index e2154bdb2..7956ce558 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -148,6 +148,7 @@ configs_derive = { path = "core/configs_derive", version =
"0.1.0" }
consensus = { path = "core/consensus" }
console-subscriber = "0.5.0"
cpu_allocation = { path = "core/cpu_allocation" }
+crc32c = "0.6.8"
crossbeam = "0.8.5"
crossfire = "3.1.20"
csv = "1.4.0"
@@ -177,6 +178,7 @@ figlet-rs = "1.0.0"
figment = { version = "0.10.19", features = ["toml", "env"] }
file-operation = "0.8.32"
flatbuffers = "25.12.19"
+flate2 = "1.1.10"
flume = "0.12.0"
fs2 = "0.4.3"
futures = "0.3.34"
@@ -225,6 +227,7 @@ lazy_static = "1.5.0"
left-right = "=0.11.8"
libc = "0.2.189"
log = "0.4.34"
+lz4 = "1.28.1"
lz4_flex = "0.14.0"
meilisearch-sdk = { version = "0.33.0", default-features = false, features = [
"reqwest",
@@ -308,6 +311,7 @@ server_common = { path = "core/server_common" }
shard = { path = "core/shard" }
simd-json = { version = "0.18.1", features = ["serde_impl"] }
smallvec = "1.16"
+snap = "1.1.2"
socket2 = "0.6.5"
sqlparser = { version = "0.63.0", features = ["visitor"] }
sqlx = { version = "0.9.0", features = [
@@ -379,6 +383,7 @@ yew-router = "0.20"
zbus-secret-service-keyring-store = { version = "1.0.1", features =
["rt-async-io-crypto-rust"] }
zeroize = "1.9.0"
zip = { version = "8.6.0", default-features = false, features = ["deflate"] }
+zstd = "0.13.3"
[profile.release]
lto = true
diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml
index f2c76e644..63bfbca6d 100644
--- a/gateways/kafka/Cargo.toml
+++ b/gateways/kafka/Cargo.toml
@@ -33,15 +33,20 @@ name = "iggy-gateway-kafka"
path = "src/main.rs"
[dependencies]
+# Required by kafka-protocol: its custom-decompression hook is typed on
anyhow::Result.
+anyhow = { workspace = true }
bytes = { workspace = true }
+# The four codecs are driven from `records::decompress` rather than through
kafka-protocol,
+# whose decompressors materialize the whole stream before a budget can see it.
+flate2 = { workspace = true }
iggy = { workspace = true }
-# Broker-role only: decodes requests and encodes responses. Default features
also pull in
-# client-role codec paths and compression codecs (gzip/lz4/snappy/zstd) this
gateway never
-# uses, since RecordBatch payloads stay opaque `Bytes` here.
-kafka-protocol = { workspace = true }
+# The codec features stay on for the encoder, which the batch tests use to
build fixtures.
+kafka-protocol = { workspace = true, features = ["gzip", "lz4", "snappy",
"zstd"] }
libc = { workspace = true }
+lz4 = { workspace = true }
secrecy = { workspace = true }
serde = { workspace = true }
+snap = { workspace = true }
socket2 = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = [
@@ -58,8 +63,11 @@ toml = { workspace = true }
tracing = { workspace = true }
tracing-appender = { workspace = true }
tracing-subscriber = { workspace = true }
+zstd = { workspace = true }
[dev-dependencies]
+# Repairs the CRC of a batch header a test rewrote, which the decoder checks
first.
+crc32c = { workspace = true }
serial_test = { workspace = true }
tempfile = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net",
"io-util", "time"] }
diff --git a/gateways/kafka/docs/BRIDGE_MAPPING.md
b/gateways/kafka/docs/BRIDGE_MAPPING.md
index da213deeb..8791feffb 100644
--- a/gateways/kafka/docs/BRIDGE_MAPPING.md
+++ b/gateways/kafka/docs/BRIDGE_MAPPING.md
@@ -35,6 +35,7 @@ Produce, per record:
| Kafka | Iggy |
| ------- | ------ |
+| (nothing) | `kafka.v` user header, one byte, the mapping version |
| record value | `payload` |
| record key | `kafka.key` user header, `Raw` |
| record header `name` | `kafka.h.<name>` user header, `Raw` |
@@ -43,11 +44,51 @@ Produce, per record:
| partition index | partition index, both 0-based |
| topic | stream and topic per `TopicMapping` |
-Fetch reverses it. A message with no `kafka.*` headers is a message an Iggy
client wrote, and
-encodes as a Kafka record with a null key, its Iggy user headers as Kafka
headers, and
-`origin_timestamp` as the record timestamp (falling back to the
server-assigned `timestamp`
-when the origin timestamp is zero). Iggy header kinds other than `Raw` and
`String` are emitted
-as their raw value bytes.
+Fetch reverses it. A message with no `kafka.v` header is a message an Iggy
client wrote. It
+encodes as a Kafka record with a null key, and its Iggy user headers become
Kafka headers under
+their own names. The record timestamp comes from `origin_timestamp`. If the
origin timestamp is zero, it comes
+from the server-assigned `timestamp` instead.
+
+An Iggy header key carries a kind as well as bytes. If the bytes are UTF-8,
they are the Kafka
+header name, whatever the kind says. The kind is the producer's choice, and a
Kafka name is a
+string either way. Iggy header values are emitted as their raw bytes.
+
+### Provenance
+
+`kafka.v` is on every message the gateway writes and on no other. Fetch reads
it first, and
+everything else in this document follows from what it says.
+
+Present and equal to the version this build implements: the gateway wrote this
message. Every
+marker on it is authoritative, and a marker this build does not recognize is
an error rather than
+a guess. Only `kafka.h.` headers reach the consumer as Kafka headers.
+
+Present and some other value: a later mapping wrote this message. Fetch
refuses it rather than
+reading it under rules that changed after it was written.
+
+Absent: an Iggy client wrote this message. No `kafka.` header on it means
anything, every header
+passes through under its own name, and the payload is the record value.
+
+The alternative was to take any `kafka.`-prefixed header as proof of
provenance. This document
+reserves the namespace and nothing in the server enforces it, so that proof
was worthless. One
+stray `kafka.` header hid every other header on a message, and a `kafka.value`
of `null` turned
+the message into a tombstone. One version header costs about 14 bytes against
the 100 KB header
+budget and settles all of it.
+
+The native path is versioned for a second reason. It is the storage form for
nearly every record,
+and open questions 2 and 3 below can still change it. Without a version on the
stored message, a
+change to either one decodes older records wrongly and gives no way to notice.
+
+What `kafka.v` does not give is enforcement. It is a claim the message makes.
The server keeps no
+namespace for it, so an Iggy writer can set `kafka.v` to a version no build
implements. Every read
+of that message then fails. Fetch cannot serve a record it cannot decode, and
a Kafka consumer
+cannot step over one. One such message therefore stalls the partition for
every Kafka consumer.
+
+This is the open end of the provenance design, and it belongs to Fetch rather
than to the mapping.
+The handler in [#3536](https://github.com/apache/iggy/issues/3536) owns the
policy for a message
+the mapping refuses. Two shapes are on the table. Skipping serves the records
around it and
+leaves a gap, which Kafka consumers already tolerate on a compacted topic. Its
cost is that a
+message goes missing with no signal. Quarantining records the offset and
surfaces a metric. Its
+cost is somewhere to keep the record. Neither is decided here.
### Timestamps
@@ -63,6 +104,25 @@ Kafka sends `-1` for a record with no timestamp. That is
stored as `0`, and Fetc
a zero origin timestamp as an instruction to use the server-assigned timestamp
instead. A real
broker does the same thing under `LogAppendTime`, so the two agree.
+A record stamped at exactly `0` ms, the Unix epoch, is a different record that
stores the same
+`0`. It carries a `kafka.ts` header holding `epoch` to say so, and Fetch reads
that header before
+it reads the origin timestamp. Without it a producer that stamps a record `0`
gets the server's
+clock back instead, and never learns that the value changed.
+
+One Iggy batch holds timestamps that span at most
`MAX_TIMESTAMP_DELTA_MICROS`, which is
+`u32::MAX` microseconds, about 71.6 minutes
(`core/binary_protocol/src/batch.rs:55`). The send
+encoder stores each message as a `u32` delta from the batch minimum and
refuses a larger one
+(`core/binary_protocol/src/requests/messages/send_messages.rs:133`). Kafka
puts no such bound on
+one produce batch, so two shapes fail:
+
+- a batch whose CreateTime values span more than 71.6 minutes, which replay
and mirror producers
+ reach
+- a batch that mixes a record with no timestamp, stored as `0`, with normally
stamped records
+
+Neither is fixable in the record mapping, which sees one record and has no
batch minimum to work
+from. Produce ([#3535](https://github.com/apache/iggy/issues/3535)) owns
splitting a produce
+batch into sends the server accepts.
+
## Records Iggy cannot hold natively
Iggy rejects an empty payload
(`core/common/src/types/message/iggy_message.rs:169`), caps a
@@ -95,12 +155,20 @@ A record takes the fallback when any of these hold:
- the key is longer than 255 bytes
- a header name, prefixed with `kafka.h.`, is longer than 255 bytes
- a header value is null, empty, or longer than 255 bytes
-- two headers share a name
- the headers together would exceed the 100 KB user-header budget
-Such a record is stored with a `kafka.envelope` header whose value is one
byte, the format
-version, currently `1`. The payload holds the key, the value and the headers
in the layout
-below. Fetch checks for that header first and takes the plain path only when
it is absent.
+A repeated header name belonged on that list and is not reachable. Kafka
allows one, but
+`kafka_protocol` decodes headers into an `IndexMap` and inserts each in turn
(`records.rs:919`),
+so a repeat overwrites its earlier entry before any gateway code runs. The
last value wins and
+the record takes the native path. Catching the case needs a decoder this
gateway does not have.
+
+Such a record is stored with a `kafka.envelope` header whose value is one
byte, the byte layout
+version, currently `1`. The payload holds the key, the value and the headers
in the layout below.
+Fetch checks for that header first and takes the plain path only when it is
absent.
+
+That byte numbers the layout below and nothing else. `kafka.v` numbers the
mapping as a whole.
+Both exist because the envelope layout can change while the rest of the
mapping stands. A record
+on the native path also needs a version, and it has no envelope to carry one.
The layout is fixed here rather than left to the implementation, because
`kafka_protocol` hands
a handler a decoded `Record` and never a raw slice of the record, so there is
no verbatim body
@@ -129,6 +197,12 @@ The cost is that these messages are opaque to Iggy
consumers and connectors. Tha
of confining the fallback to record shapes that are rare in practice, rather
than making it the
default storage form.
+The envelope moves the key and the headers into the payload, so it cannot hold
every record the
+native path cannot. A value close enough to `MAX_PAYLOAD_SIZE`, on a record
whose key is empty or
+over 255 bytes, comes to more than `MAX_PAYLOAD_SIZE` once the envelope wraps
it, and Produce
+rejects the record with `MESSAGE_TOO_LARGE` (10). The gateway measures the
envelope before it
+builds one, so the answer does not depend on allocating a payload already
known to be too big.
+
## Batch-level fields
Per-record storage drops what the Kafka record batch header carries: producer
id, producer
@@ -148,9 +222,41 @@ Whether producer id `-1` is what Fetch actually sends
depends on the InitProduce
[`IDEMPOTENCE.md`](IDEMPOTENCE.md). Allocating producer ids does not change
what is stored, only
what Produce accepts, so this section holds under either answer.
-Produce decompresses gzip, snappy, lz4 and zstd batches, which means turning
those features back
-on for the `kafka-protocol` dependency (`Cargo.toml:217` currently builds it
with
-`default-features = false, features = ["broker"]`). Fetch emits uncompressed
batches.
+Produce refuses a control batch and a transactional batch. A stored message
carries neither flag.
+A control record admitted here therefore reaches consumers as ordinary
application data, and a
+consumer filters control records by exactly that flag. Transactions are out of
scope in
+[`IDEMPOTENCE.md`](IDEMPOTENCE.md), so refusing is the answer that leaves a
consumer's view
+intact.
+
+Fetch writes one batch per response rather than one per record. The encoder
groups records while
+`offset - sequence` holds, so each record's `sequence` is numbered from the
first offset in the
+response. The record constructor leaves `sequence` at `-1` on every record.
That breaks the group
+on every record, and each one then carries its own 61-byte batch header.
+
+Two counts in a batch drive an allocation, and upstream checks each one for
sign alone. Both are
+bounded here, in different places, because no single place can see them both.
+
+The batch header's record count sits in the header, and `RecordBatchDecoder`
reserves a `Vec` from
+it (`kafka-protocol-0.18.0/src/records.rs:517`). A 61-byte batch declaring
`i32::MAX` records asks
+for 377 GB. Produce reads every batch header before it decodes a record and
refuses a count the
+blob cannot produce. That ceiling is the blob's own length. If something in
the blob is
+compressed, the decompression budget is added to it. An uncompressed batch's
records are already
+in the blob, and granting it the budget as well lets a 70-byte batch declare a
million records.
+
+The per-record header count sits inside a record body, as a varint, so no
batch header reports it
+and the header pass cannot see it. `Record::decode_new` reserves an `IndexMap`
from it
+(`kafka-protocol-0.18.0/src/records.rs:896`), and that allocation is resident
rather than virtual,
+because hashbrown writes its control bytes. A 72-byte batch declaring one
record and `i32::MAX`
+headers aborts the process. Produce therefore walks the record bytes in the
decompression step,
+which runs for every batch including an uncompressed one. It refuses a header
count the record's
+own bytes cannot hold, at two length varints per header.
+
+Neither bound changes the ratio between a minimal wire record and a decoded
`Record`. A
+legitimate request of small records pays that ratio too.
+
+Produce decompresses gzip, snappy, lz4 and zstd batches.
`gateways/kafka/Cargo.toml` turns those
+four features on for `kafka-protocol` and the workspace entry stays on
`broker` alone, so the
+codecs are declared by the crate that needs them. Fetch emits uncompressed
batches.
Decompression needs its own bound, and the bound is per request rather than
per batch.
`max_frame_size` bounds the frame a client sent, which is the compressed size,
and zstd reaches
@@ -161,13 +267,24 @@ therefore still admit 4096 times that much output.
Produce keeps a single decompression budget for the whole request, set to
`max_frame_size`, so a
compressed request can never yield more than the same client could have sent
uncompressed. A
-batch that exhausts the budget is rejected with `MESSAGE_TOO_LARGE` (10)
before the output is
-allocated. Each decompressed record value has to clear Iggy's own
`MAX_PAYLOAD_SIZE` (64 MB,
-`iggy_message.rs:44`) separately, since one record becomes one message.
+batch that exhausts the budget is rejected with `MESSAGE_TOO_LARGE` (10). Each
decompressed
+record value has to clear Iggy's own `MAX_PAYLOAD_SIZE` (64 MB,
`iggy_message.rs:44`) separately,
+since one record becomes one message.
+
+The budget bounds the peak, not just the total. `kafka_protocol`'s own
decompressors write the
+whole stream into a growing buffer before they hand it over
(`compression/gzip.rs:46` and its
+three siblings). A batch decoded through them reaches its full decompressed
size in memory, and
+the budget then reports an overrun that already happened.
+
+The gateway therefore drives the four codecs itself, through an `io::Write`
sink that refuses the
+write past the budget. Gzip, zstd and lz4 stream through that sink. Snappy is
the one codec that
+states its output size up front, and `snap` reads that number without
allocating. The declared
+size is charged before the decoder runs, for Kafka's own block framing and for
raw snappy alike.
-Nothing decompresses today. The record batch stays an opaque `Bytes` on both
paths, so the bound
-above is a requirement on [#3535](https://github.com/apache/iggy/issues/3535)
rather than a
-description of current behavior.
+`records::decode_batches` decompresses, and `records::DecompressionBudget` is
the bound. The
+budget is a parameter, so setting it to `max_frame_size` for the whole request
belongs to the
+Produce handler in [#3535](https://github.com/apache/iggy/issues/3535). Until
that lands, nothing
+calls either one in a server path.
## Offsets
@@ -194,6 +311,22 @@ carries meaning. The envelope does preserve order, since
it stores the headers a
record only reaches the envelope for one of the reasons above. A consumer that
depends on header
order therefore sees a different order through the gateway than a real broker
would give it.
+### Header name collisions
+
+An Iggy header key is bytes plus a kind, and the key orders on kind before
bytes
+(`core/common/src/types/message/user_headers.rs:209`). Two keys holding the
same bytes under
+different kinds are therefore two stored headers. A Kafka header name is a
string, so both map to
+one name.
+
+Kafka carries headers as a list and holds both. A `kafka_protocol` `Record`
keys them in an
+`IndexMap`, so nothing here keeps the pair, and the later key in stored order
wins.
+
+On a message the gateway wrote this cannot happen, because `to_iggy` writes
one key kind. Fetch
+therefore refuses such a message, rather than dropping a header from it
quietly. On a message an
+Iggy client wrote it can happen, and Fetch keeps one header and loses the
other. Refusing that
+message lets one Iggy writer stall the partition for every Kafka consumer of
it, which is the
+worse of the two.
+
## Partitioning
Both systems number partitions from 0, so the partition index passes through
unchanged in each
@@ -212,10 +345,26 @@ Iggy's group registry is used as an offset key and for
nothing else, which
## Reserved header namespace
-`kafka.` is reserved on messages the gateway writes and reads. An Iggy
producer that sets a
-header in that namespace on a topic a Kafka consumer reads will have it
interpreted as gateway
+`kafka.` is reserved on messages the gateway writes. Nothing in the server
enforces it, so the
+reservation buys nothing on its own. `kafka.v` carries the provenance instead,
as the Provenance
+section above describes. An Iggy producer that sets `kafka.value` or
`kafka.envelope` without
+`kafka.v` gets those headers back under their own names. The gateway reads
none of them as
metadata.
+A producer that sets `kafka.v` as well is claiming to be this gateway, and
Fetch takes the claim:
+a marker or an envelope that does not parse is then an error for that message.
The header block
+and the envelope payload are still treated as untrusted throughout, so the
error is a rejection
+and never an allocation. The envelope decoder charges the nine bytes each
header needs against
+the payload before it reserves for one, and it rejects a payload with bytes
left over after the
+last header.
+
+One more case is not a namespace question. Iggy folds a user-header block it
cannot parse into
+"this message has no headers"
(`core/common/src/types/message/iggy_message.rs:240`). Read that
+way, an enveloped message loses its `kafka.envelope` marker. Its envelope
bytes then reach the
+consumer as the record value, and a tombstone arrives as a one-byte message.
Fetch tells an
+unreadable block from an absent one by the stored `user_headers_length`, and
refuses the
+message.
+
## Open questions
Four questions need an answer before Produce
([#3535](https://github.com/apache/iggy/issues/3535))
@@ -239,7 +388,8 @@ Every Kafka header name is stored as `kafka.h.<name>`,
which spends 8 of the 255
header name has, on every header of every record. A shorter prefix buys those
bytes back and
costs readability for anyone reading a stream by hand.
-Default: keep `kafka.`.
+Default: keep `kafka.`. `kafka.v` makes the answer revisable: a later mapping
version can use a
+shorter prefix, and already stored records keep decoding under the version
they carry.
### 3. Is the placeholder byte acceptable for tombstones?
@@ -250,7 +400,8 @@ The alternative is the envelope, which costs a tombstone
the fast path. Tombston
traffic on compacted Kafka topics, and Iggy has no compaction, so they are
stored and served
like any other record.
-Default: keep the placeholder byte.
+Default: keep the placeholder byte. `kafka.v` makes this answer revisable too,
on the same
+terms as question 2.
### 4. Recompress on Fetch, or always emit uncompressed?
diff --git a/gateways/kafka/src/lib.rs b/gateways/kafka/src/lib.rs
index 5ccd935ae..043fe9e3d 100644
--- a/gateways/kafka/src/lib.rs
+++ b/gateways/kafka/src/lib.rs
@@ -20,6 +20,7 @@
pub mod bridge;
pub mod error;
pub mod protocol;
+pub mod records;
pub mod server;
pub use server::{GatewayConfig, KafkaGateway};
diff --git a/gateways/kafka/src/records.rs b/gateways/kafka/src/records.rs
new file mode 100644
index 000000000..950c370fe
--- /dev/null
+++ b/gateways/kafka/src/records.rs
@@ -0,0 +1,1830 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! One Kafka record to and from one Iggy message.
+//!
+//! `docs/BRIDGE_MAPPING.md` is the specification. This module implements it
and nothing else:
+//! no Iggy calls, no handler wiring.
+
+use std::cell::Cell;
+use std::collections::BTreeMap;
+use std::io::{self, Write};
+
+use bytes::{Buf, BufMut, Bytes, BytesMut};
+use iggy::prelude::{HeaderKey, HeaderValue, IggyError, IggyMessage,
MAX_PAYLOAD_SIZE};
+use kafka_protocol::indexmap::IndexMap;
+use kafka_protocol::protocol::StrBytes;
+use kafka_protocol::records::{
+ Compression, NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH, NO_PRODUCER_ID,
NO_SEQUENCE, Record,
+ RecordBatchDecoder, RecordBatchEncoder, RecordEncodeOptions, TimestampType,
+};
+use thiserror::Error;
+
+/// Iggy header whose one-byte value is the storage mapping version.
+///
+/// Written on every message this gateway produces and on no other, which is
what lets the read
+/// path tell its own messages from an Iggy client's. The `kafka.` namespace
is reserved by
+/// convention only, so presence of one namespaced header proves nothing on
its own.
+pub const VERSION_HEADER: &str = "kafka.v";
+/// Storage mapping version this build writes and reads.
+pub const MAPPING_VERSION: u8 = 1;
+
+/// Iggy header carrying the Kafka record key.
+pub const KEY_HEADER: &str = "kafka.key";
+/// Iggy header naming which of null or empty a placeholder payload stands for.
+pub const VALUE_MARKER_HEADER: &str = "kafka.value";
+/// Iggy header marking a record stamped at the Unix epoch.
+pub const TIMESTAMP_MARKER_HEADER: &str = "kafka.ts";
+/// Prefix every Kafka record header name is stored under.
+pub const HEADER_PREFIX: &str = "kafka.h.";
+/// Iggy header whose one-byte value is the envelope byte layout version.
+pub const ENVELOPE_HEADER: &str = "kafka.envelope";
+/// Envelope byte layout version this build writes and reads.
+pub const ENVELOPE_VERSION: u8 = 1;
+
+/// Kafka sends this for a record with no timestamp.
+const NO_TIMESTAMP: i64 = -1;
+/// The one Kafka timestamp an `origin_timestamp` of zero cannot be told apart
from.
+const EPOCH_TIMESTAMP: i64 = 0;
+/// Stored in place of a null or empty value, discarded on the way back.
+const PLACEHOLDER: &[u8] = &[0x00];
+/// Iggy caps one header name and one header value at this many bytes.
+const MAX_FIELD: usize = 255;
+
+const MARKER_NULL: &[u8] = b"null";
+const MARKER_EMPTY: &[u8] = b"empty";
+const MARKER_EPOCH: &[u8] = b"epoch";
+
+const FLAG_KEY: u8 = 0b01;
+const FLAG_VALUE: u8 = 0b10;
+
+/// Flags byte, key length, value length and header count, per
`BRIDGE_MAPPING.md`.
+const ENVELOPE_OVERHEAD: usize = 13;
+/// Name length, value-present byte and value length, before either field's
own bytes.
+const ENVELOPE_HEADER_OVERHEAD: usize = 9;
+
+/// Record batch version this gateway writes. v2 is the only shape
`kafka_protocol` encodes.
+const BATCH_VERSION: i8 = 2;
+
+/// Smallest v2 record: a length, an attributes byte, two deltas, two field
lengths and a header
+/// count, each a one-byte varint at least.
+const MIN_RECORD_BYTES: usize = 7;
+/// Smallest v2 record header: a name length varint and a value length varint,
both empty.
+const MIN_HEADER_BYTES: usize = 2;
+/// Bytes a zigzag varint occupies at most, which is what `kafka_protocol`
reads.
+const MAX_VARINT_BYTES: usize = 5;
+/// Bytes a zigzag varlong occupies at most, on the same terms.
+const MAX_VARLONG_BYTES: usize = 10;
+/// Base offset, batch length, leader epoch, magic, CRC, attributes, last
offset delta, first and
+/// max timestamp, producer id, producer epoch, base sequence and record count.
+const BATCH_HEADER_BYTES: usize = 61;
+/// Widest v2 record framing: five varints at five bytes each, an attributes
byte, and the header
+/// count varint, before the key, the value and the header bytes.
+const RECORD_FRAMING_BYTES: usize = 31;
+/// Widest per-header framing inside a v2 record: a name length and a value
length varint.
+const HEADER_FRAMING_BYTES: usize = 10;
+
+/// Marks a snappy stream written by Kafka's own framing rather than raw
snappy.
+///
+/// Kafka producers write xerial-framed snappy, which raw snappy decoders
reject, and the Java
+/// broker falls back to raw when the magic is absent. Both shapes therefore
reach a broker.
+const SNAPPY_MAGIC: &[u8; 16] =
b"\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01";
+
+/// Why a record or a batch could not cross.
+#[derive(Debug, Error)]
+pub enum RecordCodecError {
+ #[error("Iggy rejected the message: {0}")]
+ Iggy(#[from] IggyError),
+ #[error("record timestamp {0} ms does not fit Iggy's microsecond field")]
+ TimestampOutOfRange(i64),
+ #[error("{0} bytes of stored user headers did not parse")]
+ UserHeadersUnreadable(u32),
+ #[error("stored mapping version {0} is not {MAPPING_VERSION}")]
+ MappingVersion(u8),
+ #[error("value marker {0:?} is neither null nor empty")]
+ ValueMarker(Bytes),
+ #[error("two stored header keys both name {0}")]
+ HeaderNameCollision(String),
+ #[error("timestamp marker {0:?} is not epoch")]
+ TimestampMarker(Bytes),
+ #[error("envelope for this record is {size} bytes, over Iggy's
{MAX_PAYLOAD_SIZE} byte limit")]
+ EnvelopeTooLarge { size: usize },
+ #[error("envelope is truncated: needed {needed} bytes, {remaining}
remain")]
+ EnvelopeTruncated { needed: usize, remaining: usize },
+ #[error("envelope has {0} bytes left after its last header")]
+ EnvelopeTrailingBytes(usize),
+ #[error("envelope format version {0} is not {ENVELOPE_VERSION}")]
+ EnvelopeVersion(u8),
+ #[error("envelope header name is not UTF-8")]
+ EnvelopeHeaderName,
+ #[error("record batch is malformed: {0}")]
+ Batch(String),
+ #[error("batch declares {count} records, and {limit} bytes can hold
fewer")]
+ RecordCountTooLarge { count: i32, limit: usize },
+ #[error("record declares {count} headers, and {limit} bytes can hold
fewer")]
+ HeaderCountTooLarge { count: i32, limit: usize },
+ #[error("record batch ends inside a record")]
+ RecordTruncated,
+ #[error("record field declares {0} bytes")]
+ RecordFieldLength(i32),
+ #[error("{0} batches are out of scope")]
+ UnsupportedBatch(&'static str),
+ #[error("decompressed {produced} bytes with {remaining} left in the
request budget")]
+ BudgetExceeded { produced: usize, remaining: usize },
+}
+
+type Result<T> = std::result::Result<T, RecordCodecError>;
+
+/// Encodes one Kafka record as one Iggy message.
+///
+/// Takes the native path when Iggy can hold every field, and the envelope
otherwise. A caller
+/// cannot tell which from the return value, which is the point: `from_iggy`
reverses both.
+///
+/// # Errors
+///
+/// Returns an error when the timestamp does not fit, when the envelope would
exceed
+/// `MAX_PAYLOAD_SIZE`, or when Iggy rejects the message for a reason the
envelope does not fix.
+pub fn to_iggy(record: &Record) -> Result<IggyMessage> {
+ if needs_envelope(record) {
+ return envelope_message(record);
+ }
+ let (payload, marker) = split_value(record.value.as_ref());
+ let mut headers = gateway_headers(record.timestamp);
+ if let Some(marker) = marker {
+ headers.insert(header_key(VALUE_MARKER_HEADER), header_value(marker));
+ }
+ if let Some(key) = record.key.as_ref() {
+ headers.insert(header_key(KEY_HEADER), header_value(key));
+ }
+ for (name, value) in &record.headers {
+ // `needs_envelope` rejected the shapes that cannot be built here, so
both are infallible.
+ let Some(value) = value.as_ref() else {
+ continue;
+ };
+ headers.insert(
+ header_key(&format!("{HEADER_PREFIX}{}", name.as_str())),
+ header_value(value),
+ );
+ }
+
+ // The only limit left is the 100 KB budget over all headers together,
which no per-field
+ // check can see. Let the constructor rule on it rather than duplicating
its arithmetic.
+ build(payload, headers, record.timestamp)?.map_or_else(||
envelope_message(record), Ok)
+}
+
+/// Decodes one Iggy message as one Kafka record at `offset`.
+///
+/// A message without `kafka.v` was written by an Iggy client, not through
this gateway. It gets a
+/// null key, its own user headers under their own names, and its payload as
the record value.
+/// None of the `kafka.` headers carries meaning on such a message, because
the namespace is
+/// reserved by `BRIDGE_MAPPING.md` and by nothing the server enforces.
+///
+/// A message with `kafka.v` is taken as written here, so every marker on it
is authoritative and
+/// one this build does not recognize is an error rather than a guess. Nothing
on the server
+/// enforces the namespace, so that is a claim the message makes and not one
the server keeps: an
+/// Iggy writer can set `kafka.v` to a version this build does not implement,
and every read of
+/// that message then fails. Fetch cannot serve a record it cannot decode and
a Kafka consumer
+/// cannot step over one, so the handler in #3536 owns the skip-or-quarantine
policy for a message
+/// that fails here. `BRIDGE_MAPPING.md` records that as the open end of the
provenance design.
+///
+/// # Errors
+///
+/// Returns an error when the stored user headers do not parse, when `kafka.v`
names a mapping
+/// version this build does not implement, or when a marker or an envelope is
malformed.
+pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> {
+ let stored = user_headers(message)?;
+ let Some(version) = stored.get(&header_key(VERSION_HEADER)) else {
+ let (key, value, headers) = foreign_fields(message, &stored);
+ return Ok(record(key, value, headers, offset, timestamp_out(message)));
+ };
+ if version.as_bytes() != [MAPPING_VERSION] {
+ let version = version.as_bytes().first().copied().unwrap_or_default();
+ return Err(RecordCodecError::MappingVersion(version));
+ }
+
+ let (key, value, headers) = match stored.get(&header_key(ENVELOPE_HEADER))
{
+ Some(envelope) => decode_envelope(envelope.as_bytes(),
&message.payload)?,
+ None => gateway_fields(message, &stored)?,
+ };
+ let timestamp = match stored.get(&header_key(TIMESTAMP_MARKER_HEADER)) {
+ None => timestamp_out(message),
+ Some(marker) => match marker.as_bytes() {
+ MARKER_EPOCH => EPOCH_TIMESTAMP,
+ _ => return Err(RecordCodecError::TimestampMarker(marker.value())),
+ },
+ };
+ Ok(record(key, value, headers, offset, timestamp))
+}
+
+/// The stored user headers, with an unreadable block told apart from an
absent one.
+///
+/// `IggyMessage::user_headers_map` folds a header block it cannot parse into
`Ok(None)`, which
+/// reads the same as a message that carries no headers at all. Taken at face
value that turns an
+/// enveloped message into its own envelope bytes served as the record value.
+fn user_headers(message: &IggyMessage) -> Result<BTreeMap<HeaderKey,
HeaderValue>> {
+ match message.user_headers_map()? {
+ Some(stored) => Ok(stored),
+ None if message.header.user_headers_length > 0 => Err(
+
RecordCodecError::UserHeadersUnreadable(message.header.user_headers_length),
+ ),
+ None => Ok(BTreeMap::new()),
+ }
+}
+
+/// Kafka counts milliseconds, Iggy counts microseconds, and `-1` means the
broker assigns one.
+fn timestamp_in(millis: i64) -> Result<u64> {
+ if millis == NO_TIMESTAMP {
+ return Ok(0);
+ }
+ millis
+ .checked_mul(1000)
+ .and_then(|micros| u64::try_from(micros).ok())
+ .ok_or(RecordCodecError::TimestampOutOfRange(millis))
+}
+
+/// Zero means the producer sent no timestamp, so the server-assigned one
stands in.
+///
+/// A record stamped at the epoch stores that same zero, and `from_iggy` reads
the `kafka.ts`
+/// marker before it calls this, because Iggy has no other way to hold the
difference.
+fn timestamp_out(message: &IggyMessage) -> i64 {
+ let micros = if message.header.origin_timestamp == 0 {
+ message.header.timestamp
+ } else {
+ message.header.origin_timestamp
+ };
+ i64::try_from(micros / 1000).unwrap_or(NO_TIMESTAMP)
+}
+
+/// Whether any field of `record` is one Iggy refuses to hold natively.
+///
+/// A repeated header name is on the list in `BRIDGE_MAPPING.md` and is absent
here, because
+/// `kafka_protocol` decodes headers into an `IndexMap` (`records.rs:919`). A
repeat overwrites
+/// its earlier entry before this code runs, so the case cannot be observed.
+fn needs_envelope(record: &Record) -> bool {
+ let key_unholdable = record
+ .key
+ .as_ref()
+ .is_some_and(|key| key.is_empty() || key.len() > MAX_FIELD);
+ if key_unholdable {
+ return true;
+ }
+ record.headers.iter().any(|(name, value)| {
+ HEADER_PREFIX.len() + name.as_str().len() > MAX_FIELD
+ || value
+ .as_ref()
+ .is_none_or(|value| value.is_empty() || value.len() >
MAX_FIELD)
+ })
+}
+
+/// Payload to store, and the marker naming what the original value was when
it is not the payload.
+fn split_value(value: Option<&Bytes>) -> (Bytes, Option<&'static [u8]>) {
+ match value {
+ None => (Bytes::from_static(PLACEHOLDER), Some(MARKER_NULL)),
+ Some(value) if value.is_empty() => (Bytes::from_static(PLACEHOLDER),
Some(MARKER_EMPTY)),
+ Some(value) => (value.clone(), None),
+ }
+}
+
+/// The headers every gateway-written message carries, whichever path it takes.
+///
+/// Iggy reads an `origin_timestamp` of zero as no timestamp at all, so a
record that really was
+/// stamped at the epoch needs a marker to hold the difference.
+fn gateway_headers(timestamp: i64) -> BTreeMap<HeaderKey, HeaderValue> {
+ let mut headers = BTreeMap::new();
+ headers.insert(header_key(VERSION_HEADER),
header_value(&[MAPPING_VERSION]));
+ if timestamp == EPOCH_TIMESTAMP {
+ headers.insert(
+ header_key(TIMESTAMP_MARKER_HEADER),
+ header_value(MARKER_EPOCH),
+ );
+ }
+ headers
+}
+
+/// `Ok(None)` when the headers together pass Iggy's budget, which the
envelope then carries.
+fn build(
+ payload: Bytes,
+ headers: BTreeMap<HeaderKey, HeaderValue>,
+ timestamp: i64,
+) -> Result<Option<IggyMessage>> {
+ let mut message = match IggyMessage::builder()
+ .payload(payload)
+ .user_headers(headers)
+ .build()
+ {
+ Ok(message) => message,
+ Err(IggyError::TooBigUserHeaders) => return Ok(None),
+ Err(error) => return Err(error.into()),
+ };
+ message.header.origin_timestamp = timestamp_in(timestamp)?;
+ Ok(Some(message))
+}
+
+fn envelope_message(record: &Record) -> Result<IggyMessage> {
+ // The envelope moves the key and the headers into the payload, so a
record whose value alone
+ // clears `MAX_PAYLOAD_SIZE` can be one the fallback cannot hold. Say so
before spending the
+ // allocation, since the native path has already been ruled out and
nothing else is left.
+ let size = envelope_size(record);
+ if size > MAX_PAYLOAD_SIZE as usize {
+ return Err(RecordCodecError::EnvelopeTooLarge { size });
+ }
+
+ let mut headers = gateway_headers(record.timestamp);
+ headers.insert(
+ header_key(ENVELOPE_HEADER),
+ header_value(&[ENVELOPE_VERSION]),
+ );
+ build(encode_envelope(record, size), headers, record.timestamp)?
+ .ok_or(IggyError::TooBigUserHeaders)
+ .map_err(Into::into)
+}
+
+/// Exactly what `encode_envelope` writes for `record`.
+fn envelope_size(record: &Record) -> usize {
+ let field = |field: Option<&Bytes>| field.map_or(0, Bytes::len);
+ ENVELOPE_OVERHEAD
+ + field(record.key.as_ref())
+ + field(record.value.as_ref())
+ + record
+ .headers
+ .iter()
+ .map(|(name, value)| {
+ ENVELOPE_HEADER_OVERHEAD + name.as_str().len() +
field(value.as_ref())
+ })
+ .sum::<usize>()
+}
+
+/// 13 bytes of fixed overhead plus 9 per header, little-endian throughout.
+fn encode_envelope(record: &Record, size: usize) -> Bytes {
+ let mut flags = 0u8;
+ if record.key.is_some() {
+ flags |= FLAG_KEY;
+ }
+ if record.value.is_some() {
+ flags |= FLAG_VALUE;
+ }
+
+ let mut buf = BytesMut::with_capacity(size);
+ buf.put_u8(flags);
+ put_field(&mut buf, record.key.as_ref());
+ put_field(&mut buf, record.value.as_ref());
+ buf.put_u32_le(u32::try_from(record.headers.len()).unwrap_or(u32::MAX));
+ for (name, value) in &record.headers {
+ let name = name.as_str().as_bytes();
+ buf.put_u32_le(u32::try_from(name.len()).unwrap_or(u32::MAX));
+ buf.put_slice(name);
+ buf.put_u8(u8::from(value.is_some()));
+ put_field(&mut buf, value.as_ref());
+ }
+ buf.freeze()
+}
+
+type RecordFields = (
+ Option<Bytes>,
+ Option<Bytes>,
+ IndexMap<StrBytes, Option<Bytes>>,
+);
+
+fn decode_envelope(version: &[u8], payload: &Bytes) -> Result<RecordFields> {
+ match version.first() {
+ Some(&ENVELOPE_VERSION) => {}
+ Some(&other) => return Err(RecordCodecError::EnvelopeVersion(other)),
+ None => return Err(RecordCodecError::EnvelopeVersion(0)),
+ }
+
+ let mut buf = payload.clone();
+ let flags = take(&mut buf, 1)?[0];
+ let key = take_field(&mut buf)?;
+ let value = take_field(&mut buf)?;
+ let count =
+ u32::from_le_bytes(take(&mut buf,
4)?.as_ref().try_into().unwrap_or_default()) as usize;
+
+ // The count is four bytes of a payload anyone can write and every header
costs at least nine,
+ // so reserving before reading lets a 13-byte message ask for four billion
entries. Charge the
+ // floor against what is left and the reserve below is bounded by the
input.
+ let needed = count.saturating_mul(ENVELOPE_HEADER_OVERHEAD);
+ if buf.remaining() < needed {
+ return Err(RecordCodecError::EnvelopeTruncated {
+ needed,
+ remaining: buf.remaining(),
+ });
+ }
+
+ let mut headers = IndexMap::with_capacity(count);
+ for _ in 0..count {
+ let name = take_field(&mut buf)?;
+ let name =
+ String::from_utf8(name.to_vec()).map_err(|_|
RecordCodecError::EnvelopeHeaderName)?;
+ let present = take(&mut buf, 1)?[0] != 0;
+ let value = take_field(&mut buf)?;
+ headers.insert(StrBytes::from_string(name), present.then_some(value));
+ }
+
+ // Every byte of an envelope is accounted for above, so a leftover means
this payload is not
+ // one. Accepting it would turn a stray `kafka.envelope` header plus junk
into a null record.
+ if buf.has_remaining() {
+ return Err(RecordCodecError::EnvelopeTrailingBytes(buf.remaining()));
+ }
+
+ Ok((
+ (flags & FLAG_KEY != 0).then_some(key),
+ (flags & FLAG_VALUE != 0).then_some(value),
+ headers,
+ ))
+}
+
+/// The native reading of a message this gateway wrote, so every marker on it
is authoritative.
+fn gateway_fields(
+ message: &IggyMessage,
+ stored: &BTreeMap<HeaderKey, HeaderValue>,
+) -> Result<RecordFields> {
+ let key = stored.get(&header_key(KEY_HEADER)).map(HeaderValue::value);
+ let value = match stored.get(&header_key(VALUE_MARKER_HEADER)) {
+ None => Some(message.payload.clone()),
+ Some(marker) => match marker.as_bytes() {
+ MARKER_NULL => None,
+ MARKER_EMPTY => Some(Bytes::new()),
+ // A marker this build does not write, on a message that says this
build wrote it.
+ // Guessing loses a payload or invents one, and a later version
that adds a third
+ // marker is read wrongly here rather than refused.
+ _ => return Err(RecordCodecError::ValueMarker(marker.value())),
+ },
+ };
+
+ let mut headers = IndexMap::new();
+ for (name, stored_value) in stored {
+ let Some(name) = header_name(name).and_then(|name|
name.strip_prefix(HEADER_PREFIX)) else {
+ continue;
+ };
+ // Two stored keys can hold the same bytes under different kinds,
because an Iggy header
+ // key orders on kind before bytes (`user_headers.rs:209`). `to_iggy`
writes one kind, so
+ // on a message this build wrote the names cannot collide, and a
collision here says the
+ // message is not what its `kafka.v` claims. Overwriting would drop a
header quietly.
+ let name = StrBytes::from_string(name.to_string());
+ if headers
+ .insert(name.clone(), Some(stored_value.value()))
+ .is_some()
+ {
+ return
Err(RecordCodecError::HeaderNameCollision(name.to_string()));
+ }
+ }
+ Ok((key, value, headers))
+}
+
+/// The reading of a message an Iggy client wrote, which no marker on it can
change.
+///
+/// Every header passes through under its own name, including one in the
`kafka.` namespace. The
+/// alternative was to treat any namespaced header as gateway metadata, which
made a single
+/// `kafka.`-prefixed header hide every other header on the message.
+///
+/// Two keys holding the same bytes under different kinds are two stored
headers and one Kafka
+/// header, and the later one in key order wins. Kafka carries headers as a
list and would hold
+/// both, but a `Record` keys them in an `IndexMap`, so there is no shape here
that keeps the
+/// pair. Refusing the message instead would let one Iggy writer stall a
partition for every
+/// Kafka consumer of it, which is the worse of the two. `BRIDGE_MAPPING.md`
records the loss.
+fn foreign_fields(
+ message: &IggyMessage,
+ stored: &BTreeMap<HeaderKey, HeaderValue>,
+) -> RecordFields {
+ let mut headers = IndexMap::new();
+ for (name, stored_value) in stored {
+ let Some(name) = header_name(name) else {
+ continue;
+ };
+ headers.insert(
+ StrBytes::from_string(name.to_string()),
+ Some(stored_value.value()),
+ );
+ }
+ (None, Some(message.payload.clone()), headers)
+}
+
+/// A Kafka header name for an Iggy header key, or `None` when the key is not
text.
+///
+/// An Iggy key is bytes plus a kind, and `HeaderField::as_str` refuses every
kind but `String`
+/// (`user_headers.rs:372`). Other SDKs hand out `Raw` and `Int32` key
constructors, so a key
+/// holding a perfectly good Kafka name arrives under a kind this one would
reject. Read the bytes
+/// and let UTF-8 decide.
+fn header_name(key: &HeaderKey) -> Option<&str> {
+ std::str::from_utf8(key.as_bytes()).ok()
+}
+
+const fn record(
+ key: Option<Bytes>,
+ value: Option<Bytes>,
+ headers: IndexMap<StrBytes, Option<Bytes>>,
+ offset: i64,
+ timestamp: i64,
+) -> Record {
+ Record {
+ transactional: false,
+ control: false,
+ delete_horizon: false,
+ partition_leader_epoch: NO_PARTITION_LEADER_EPOCH,
+ producer_id: NO_PRODUCER_ID,
+ producer_epoch: NO_PRODUCER_EPOCH,
+ timestamp_type: TimestampType::Creation,
+ offset,
+ sequence: NO_SEQUENCE,
+ timestamp,
+ key,
+ value,
+ headers,
+ }
+}
+
+fn put_field(buf: &mut BytesMut, field: Option<&Bytes>) {
+ let field = field.map_or(&[][..], |field| field.as_ref());
+ buf.put_u32_le(u32::try_from(field.len()).unwrap_or(u32::MAX));
+ buf.put_slice(field);
+}
+
+fn take(buf: &mut Bytes, needed: usize) -> Result<Bytes> {
+ if buf.remaining() < needed {
+ return Err(RecordCodecError::EnvelopeTruncated {
+ needed,
+ remaining: buf.remaining(),
+ });
+ }
+ Ok(buf.split_to(needed))
+}
+
+/// Length-prefixed bytes, possibly empty. Presence is the flags byte's job,
not the length's,
+/// so that an empty key stays distinct from a null one.
+fn take_field(buf: &mut Bytes) -> Result<Bytes> {
+ let len = u32::from_le_bytes(take(buf,
4)?.as_ref().try_into().unwrap_or_default()) as usize;
+ take(buf, len)
+}
+
+/// Both are infallible for the names and values this module builds: every one
is non-empty and
+/// within `MAX_FIELD`, which `needs_envelope` guarantees for caller-supplied
bytes.
+fn header_key(name: &str) -> HeaderKey {
+ HeaderKey::try_from(name).unwrap_or_else(|_| unreachable!("header name
{name} is out of range"))
+}
+
+fn header_value(value: &[u8]) -> HeaderValue {
+ HeaderValue::try_from(value).unwrap_or_else(|_| unreachable!("header value
is out of range"))
+}
+
+/// What one Produce request may decompress to, in total.
+///
+/// Charged across every batch in the request, because one frame carries many
batches and a cap
+/// applied to each on its own admits as many multiples of it as the frame
holds entries.
+///
+/// The budget bounds the peak, not just the total. Every decompressor here
writes through
+/// `BudgetedWriter`, which refuses the write that would pass the budget, so
an over-budget frame
+/// never reaches its full decompressed size in memory.
+pub struct DecompressionBudget {
+ remaining: Cell<usize>,
+ /// Why this module refused a batch, when it did. Everything this module
reports from inside
+ /// the decoder's decompression hook leaves as an `io::Error` or an
`anyhow::Error`, which the
+ /// decoder stringifies, so the typed reason is parked here and taken by
`decode_batches`.
+ reason: Cell<Option<RecordCodecError>>,
+}
+
+impl DecompressionBudget {
+ #[must_use]
+ pub const fn new(bytes: usize) -> Self {
+ Self {
+ remaining: Cell::new(bytes),
+ reason: Cell::new(None),
+ }
+ }
+
+ fn charge(&self, produced: usize) -> io::Result<()> {
+ let remaining = self.remaining.get();
+ if produced > remaining {
+ self.refuse(RecordCodecError::BudgetExceeded {
+ produced,
+ remaining,
+ });
+ return Err(io::Error::other(format!(
+ "decompressed {produced} bytes with {remaining} left in the
budget"
+ )));
+ }
+ self.remaining.set(remaining - produced);
+ Ok(())
+ }
+
+ /// Parks the typed reason for the error about to be raised through the
hook.
+ fn refuse(&self, reason: RecordCodecError) {
+ self.reason.set(Some(reason));
+ }
+
+ /// The typed reason for a decoder error, when this module is what caused
it.
+ ///
+ /// Taken rather than read, because the budget outlives one batch. Left in
place, the first
+ /// refusal would reclassify every later error on the same request as that
same refusal.
+ fn reason(&self, error: &str) -> RecordCodecError {
+ self.reason
+ .take()
+ .unwrap_or_else(|| RecordCodecError::Batch(error.to_string()))
+ }
+}
+
+/// An `io::Write` sink that stops at the budget rather than after it.
+///
+/// Charging the output once it exists is too late: the decompressor has
already allocated it.
+struct BudgetedWriter<'a> {
+ out: BytesMut,
+ budget: &'a DecompressionBudget,
+}
+
+impl<'a> BudgetedWriter<'a> {
+ fn new(budget: &'a DecompressionBudget) -> Self {
+ Self {
+ out: BytesMut::new(),
+ budget,
+ }
+ }
+}
+
+impl Write for BudgetedWriter<'_> {
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ self.budget.charge(buf.len())?;
+ self.out.extend_from_slice(buf);
+ Ok(buf.len())
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ Ok(())
+ }
+}
+
+/// Decodes every record batch a Produce partition entry carries.
+///
+/// A partition's `records` field is one blob that holds one or more batches
back to back, so
+/// this drains `buf` rather than reading a single batch.
+///
+/// # Errors
+///
+/// Returns an error when a batch is malformed, when it declares more records
than the request
+/// can hold, when it is a control or transactional batch, or when the request
decompresses to
+/// more than `budget` allows.
+pub fn decode_batches(buf: &mut Bytes, budget: &DecompressionBudget) ->
Result<Vec<Record>> {
+ preflight(buf, budget)?;
+
+ let mut records = Vec::new();
+ while buf.has_remaining() {
+ let set = RecordBatchDecoder::decode_with_custom_compression(
+ buf,
+ Some(|compressed: &mut Bytes, compression| decompress(compressed,
compression, budget)),
+ )
+ .map_err(|error| budget.reason(&error.to_string()))?;
+ records.extend(set.records);
+ }
+ Ok(records)
+}
+
+/// Reads the batch headers before anything decodes a record.
+///
+/// `RecordBatchDecoder` reserves from the batch header's record count before
it reads the first
+/// record (`kafka-protocol-0.18.0/src/records.rs:517`), and that count is
checked for sign only.
+/// A 61-byte batch can therefore ask for `i32::MAX` records. The counts are
charged against what
+/// the blob can produce, together rather than one batch at a time, because a
blob holds many
+/// batches and each reserve lands in the same `Vec`. Reading the headers
costs one `Bytes` clone,
+/// which is a refcount.
+///
+/// What the blob can produce is its own length, and the decompression budget
on top only when
+/// something in it is compressed. An uncompressed batch's records are already
in the blob, so
+/// granting it the budget as well would let a 70-byte batch declare a million
records.
+///
+/// Control and transactional batches are refused in the same pass. `record()`
cannot carry either
+/// flag, so a control batch admitted here would reach consumers as ordinary
data, and consumers
+/// filter control records by exactly that flag.
+fn preflight(buf: &Bytes, budget: &DecompressionBudget) -> Result<()> {
+ let mut headers = buf.clone();
+ let infos = RecordBatchDecoder::decode_batch_info(&mut headers)
+ .map_err(|error| RecordCodecError::Batch(error.to_string()))?;
+
+ let compressed = infos
+ .iter()
+ .any(|info| info.compression != Compression::None);
+ let limit = if compressed {
+ buf.len().saturating_add(budget.remaining.get())
+ } else {
+ buf.len()
+ };
+
+ let mut declared = 0usize;
+ for info in &infos {
+ if info.transactional {
+ return Err(RecordCodecError::UnsupportedBatch("transactional"));
+ }
+ if info.control {
+ return Err(RecordCodecError::UnsupportedBatch("control"));
+ }
+ declared =
declared.saturating_add(usize::try_from(info.record_count).unwrap_or(0));
+ if declared.saturating_mul(MIN_RECORD_BYTES) > limit {
+ return Err(RecordCodecError::RecordCountTooLarge {
+ count: info.record_count,
+ limit,
+ });
+ }
+ }
+ Ok(())
+}
+
+/// Encodes records as one uncompressed v2 batch.
+///
+/// Fetch always emits uncompressed, so the read path spends no CPU on a codec
the client did
+/// not ask for. `BRIDGE_MAPPING.md` records that as a default open to
revisiting.
+///
+/// Takes the records by mutable reference to normalize `sequence`. The
encoder groups records
+/// while `offset - sequence` holds
(`kafka-protocol-0.18.0/src/records.rs:277`), and `record()`
+/// pins `sequence` at `NO_SEQUENCE` while offsets advance, which breaks the
group on every record
+/// and spends a 61-byte batch header on each one. Numbering each sequence
from the first offset
+/// keeps one batch and leaves the encoded `base_sequence` at `NO_SEQUENCE`.
Fetch reads a
+/// partition forward, so the first record carries the lowest offset, and that
is what makes the
+/// encoded `base_sequence` come out at `NO_SEQUENCE` rather than at an offset.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the batch.
+pub fn encode_batch(records: &mut [Record]) -> Result<Bytes> {
+ let Some(base) = records.first().map(|record| record.offset) else {
+ return Ok(Bytes::new());
+ };
+ for record in records.iter_mut() {
+ let delta = record.offset.saturating_sub(base);
+ record.sequence = i32::try_from(delta)
+ .unwrap_or(i32::MAX)
+ .saturating_add(NO_SEQUENCE);
+ }
+
+ let mut buf = BytesMut::with_capacity(batch_size(records));
+ let options = RecordEncodeOptions {
+ version: BATCH_VERSION,
+ compression: Compression::None,
+ };
+ RecordBatchEncoder::encode(&mut buf, records.iter(), &options)
+ .map_err(|error| RecordCodecError::Batch(error.to_string()))?;
+ Ok(buf.freeze())
+}
+
+/// An upper bound on the encoded batch, so the Fetch path reserves once
instead of doubling.
+///
+/// `kafka_protocol` never reserves: it writes every field with `put_slice`
into whatever buffer
+/// it is handed. Every length needed here is already in hand.
+fn batch_size(records: &[Record]) -> usize {
+ let field = |field: Option<&Bytes>| field.map_or(0, Bytes::len);
+ BATCH_HEADER_BYTES
+ + records
+ .iter()
+ .map(|record| {
+ RECORD_FRAMING_BYTES
+ + field(record.key.as_ref())
+ + field(record.value.as_ref())
+ + record
+ .headers
+ .iter()
+ .map(|(name, value)| {
+ HEADER_FRAMING_BYTES + name.as_str().len() +
field(value.as_ref())
+ })
+ .sum::<usize>()
+ })
+ .sum::<usize>()
+}
+
+/// Decompresses one batch, refusing the write that would pass the request
budget.
+///
+/// `kafka_protocol`'s own decompressors write the whole stream into a growing
buffer before they
+/// hand it over, so the four codecs are driven from here instead. Each one
writes through
+/// `BudgetedWriter`, which bounds the peak rather than reporting an overrun
after the fact.
+///
+/// The decoder calls this for every batch, uncompressed ones included, which
is what makes it the
+/// one place that sees the record bytes before anything reserves from what
they declare.
+fn decompress(
+ compressed: &mut Bytes,
+ compression: Compression,
+ budget: &DecompressionBudget,
+) -> anyhow::Result<Bytes> {
+ let body = compressed.copy_to_bytes(compressed.remaining());
+ let mut writer = BudgetedWriter::new(budget);
+ let records = match compression {
+ Compression::None => {
+ budget.charge(body.len())?;
+ body
+ }
+ Compression::Gzip => {
+ let mut decoder = flate2::write::GzDecoder::new(&mut writer);
+ decoder.write_all(&body)?;
+ decoder.finish()?;
+ writer.out.freeze()
+ }
+ Compression::Zstd => {
+ zstd::stream::copy_decode(body.as_ref(), &mut writer)?;
+ writer.out.freeze()
+ }
+ Compression::Lz4 => {
+ let mut decoder = lz4::Decoder::new(body.as_ref())?;
+ io::copy(&mut decoder, &mut writer)?;
+ decoder.finish().1?;
+ writer.out.freeze()
+ }
+ Compression::Snappy => {
+ inflate_snappy(&body, &mut writer)?;
+ writer.out.freeze()
+ }
+ };
+
+ if let Err(reason) = scan_records(&records) {
+ let message = reason.to_string();
+ budget.refuse(reason);
+ anyhow::bail!(message);
+ }
+ Ok(records)
+}
+
+/// Walks the records of one batch and refuses a header count the record
cannot hold.
+///
+/// `kafka_protocol` reserves an `IndexMap` from each record's own header count
+/// (`kafka-protocol-0.18.0/src/records.rs:896`), which is checked there for
sign alone. That
+/// count is a varint inside a record body, so no batch header reports it and
`preflight` cannot
+/// see it. A 72-byte batch declaring `i32::MAX` headers on its one record
therefore reaches the
+/// reserve, and that allocation is resident rather than virtual, because
hashbrown writes its
+/// control bytes. Every header costs two length varints at least, so a count
past half the
+/// record's remaining bytes is refused before the decoder reads the record.
+///
+/// The framing below mirrors `Record::decode_new`, field for field, so a
record this refuses is
+/// one the decoder would refuse too. The one place the two differ is trailing
bytes: the decoder
+/// stops after the count the batch header gave and ignores anything after it,
while this walks to
+/// the end of the blob. A v2 batch carries its records and nothing else, so
there is nothing to
+/// ignore.
+fn scan_records(records: &Bytes) -> Result<()> {
+ let mut blob = records.as_ref();
+ while !blob.is_empty() {
+ let size = take_varint(&mut blob)?;
+ let size = usize::try_from(size).map_err(|_|
RecordCodecError::RecordFieldLength(size))?;
+ let mut record = take_bytes(&mut blob, size)?;
+
+ take_bytes(&mut record, 1)?; // attributes
+ skip_varint(&mut record, MAX_VARLONG_BYTES)?; // timestamp delta
+ skip_varint(&mut record, MAX_VARINT_BYTES)?; // offset delta
+ skip_field(&mut record)?; // key
+ skip_field(&mut record)?; // value
+
+ let count = take_varint(&mut record)?;
+ let holds = usize::try_from(count)
+ .map_err(|_| RecordCodecError::HeaderCountTooLarge {
+ count,
+ limit: record.len(),
+ })?
+ .saturating_mul(MIN_HEADER_BYTES);
+ if holds > record.len() {
+ return Err(RecordCodecError::HeaderCountTooLarge {
+ count,
+ limit: record.len(),
+ });
+ }
+ }
+ Ok(())
+}
+
+/// Reads a zigzag varint, five bytes at most, the way `kafka_protocol` reads
one.
+fn take_varint(buf: &mut &[u8]) -> Result<i32> {
+ let mut value = 0u32;
+ for index in 0..MAX_VARINT_BYTES {
+ let byte = take_bytes(buf, 1)?[0];
+ value |= u32::from(byte & 0x7f) << (index * 7);
+ if byte < 0x80 {
+ break;
+ }
+ }
+ Ok((value >> 1).cast_signed() ^ -(value & 1).cast_signed())
+}
+
+/// Steps over a zigzag varint of up to `most` bytes, for the fields the scan
does not read.
+fn skip_varint(buf: &mut &[u8], most: usize) -> Result<()> {
+ for _ in 0..most {
+ if take_bytes(buf, 1)?[0] < 0x80 {
+ break;
+ }
+ }
+ Ok(())
+}
+
+fn take_bytes<'a>(buf: &mut &'a [u8], len: usize) -> Result<&'a [u8]> {
+ let (head, rest) = buf
+ .split_at_checked(len)
+ .ok_or(RecordCodecError::RecordTruncated)?;
+ *buf = rest;
+ Ok(head)
+}
+
+/// A length-prefixed record field, where `-1` is the absent one Kafka writes
for a null.
+fn skip_field(buf: &mut &[u8]) -> Result<()> {
+ let len = take_varint(buf)?;
+ if len < -1 {
+ return Err(RecordCodecError::RecordFieldLength(len));
+ }
+ if len > 0 {
+ take_bytes(buf, usize::try_from(len).unwrap_or_default())?;
+ }
+ Ok(())
+}
+
+/// Kafka's snappy, and raw snappy for the producers that send it.
+///
+/// Snappy is the one codec that states its output size up front, which `snap`
reads without
+/// allocating. Charge that number before the decoder runs, because the
decoder needs the whole
+/// block laid out to write into and cannot be fed a bounded sink.
+fn inflate_snappy(body: &Bytes, writer: &mut BudgetedWriter<'_>) ->
anyhow::Result<()> {
+ let mut decoder = snap::raw::Decoder::new();
+ let mut inflate = |block: &[u8], writer: &mut BudgetedWriter<'_>| ->
anyhow::Result<()> {
+ let declared = snap::raw::decompress_len(block)?;
+ writer.budget.charge(declared)?;
+ let start = writer.out.len();
+ writer.out.resize(start.saturating_add(declared), 0);
+ decoder.decompress(block, &mut writer.out[start..])?;
+ Ok(())
+ };
+
+ let Some(mut blocks) = body.strip_prefix(SNAPPY_MAGIC) else {
+ return inflate(body.as_ref(), writer);
+ };
+ while !blocks.is_empty() {
+ let (length, rest) = blocks
+ .split_at_checked(4)
+ .ok_or_else(|| anyhow::anyhow!("snappy block length is
truncated"))?;
+ let length = u32::from_be_bytes(length.try_into()?) as usize;
+ let (block, rest) = rest
+ .split_at_checked(length)
+ .ok_or_else(|| anyhow::anyhow!("snappy block of {length} bytes is
truncated"))?;
+ inflate(block, writer)?;
+ blocks = rest;
+ }
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy::prelude::HeaderKind;
+
+ const CREATE_TIME: i64 = 1_700_000_000_123;
+
+ fn record_with(
+ key: Option<&[u8]>,
+ value: Option<&[u8]>,
+ headers: &[(&str, Option<&[u8]>)],
+ ) -> Record {
+ let headers = headers
+ .iter()
+ .map(|(name, value)| {
+ (
+ StrBytes::from_string((*name).to_string()),
+ value.map(Bytes::copy_from_slice),
+ )
+ })
+ .collect();
+ record(
+ key.map(Bytes::copy_from_slice),
+ value.map(Bytes::copy_from_slice),
+ headers,
+ 0,
+ CREATE_TIME,
+ )
+ }
+
+ fn record_at(timestamp: i64) -> Record {
+ record(
+ Some(Bytes::from_static(b"k")),
+ Some(Bytes::from_static(b"v")),
+ IndexMap::new(),
+ 0,
+ timestamp,
+ )
+ }
+
+ fn message_with(payload: &'static [u8], headers: &[(&str, &[u8])]) ->
IggyMessage {
+ let headers = headers
+ .iter()
+ .map(|(name, value)| (header_key(name), header_value(value)))
+ .collect();
+ IggyMessage::builder()
+ .payload(Bytes::from_static(payload))
+ .user_headers(headers)
+ .build()
+ .unwrap()
+ }
+
+ /// The same, plus the version header that marks a message as
gateway-written.
+ fn gateway_message(payload: &'static [u8], headers: &[(&str, &[u8])]) ->
IggyMessage {
+ let mut all = vec![(VERSION_HEADER, &[MAPPING_VERSION][..])];
+ all.extend_from_slice(headers);
+ message_with(payload, &all)
+ }
+
+ fn envelope_bytes(count: u32, trailing: &[u8]) -> Bytes {
+ let mut payload = BytesMut::new();
+ payload.put_u8(0);
+ payload.put_u32_le(0);
+ payload.put_u32_le(0);
+ payload.put_u32_le(count);
+ payload.put_slice(trailing);
+ payload.freeze()
+ }
+
+ fn is_enveloped(message: &IggyMessage) -> bool {
+ message
+ .user_headers_map()
+ .unwrap()
+ .unwrap_or_default()
+ .contains_key(&header_key(ENVELOPE_HEADER))
+ }
+
+ #[test]
+ fn
given_a_plain_record_when_round_tripped_should_keep_key_value_and_headers() {
+ let original = record_with(Some(b"k"), Some(b"v"), &[("trace",
Some(b"abc"))]);
+ let message = to_iggy(&original).unwrap();
+ assert!(!is_enveloped(&message));
+ assert_eq!(message.payload.as_ref(), b"v");
+
+ let back = from_iggy(&message, 7).unwrap();
+ assert_eq!(back.key.as_deref(), Some(&b"k"[..]));
+ assert_eq!(back.value.as_deref(), Some(&b"v"[..]));
+ assert_eq!(back.offset, 7);
+ assert_eq!(
+ back.headers.get(&StrBytes::from_static_str("trace")),
+ Some(&Some(Bytes::from_static(b"abc")))
+ );
+ assert_eq!(back.headers.len(), 1, "no gateway header reaches Kafka");
+ }
+
+ #[test]
+ fn given_a_null_value_when_round_tripped_should_stay_null() {
+ let message = to_iggy(&record_with(Some(b"k"), None, &[])).unwrap();
+ assert!(
+ !is_enveloped(&message),
+ "a tombstone stays on the fast path"
+ );
+ assert_eq!(message.payload.as_ref(), PLACEHOLDER);
+ assert_eq!(from_iggy(&message, 0).unwrap().value, None);
+ }
+
+ #[test]
+ fn
given_an_empty_value_when_round_tripped_should_stay_empty_and_not_null() {
+ let message = to_iggy(&record_with(Some(b"k"), Some(b""),
&[])).unwrap();
+ assert_eq!(message.payload.as_ref(), PLACEHOLDER);
+ assert_eq!(
+ from_iggy(&message, 0).unwrap().value.as_deref(),
+ Some(&[][..])
+ );
+ }
+
+ #[test]
+ fn given_an_empty_key_when_stored_should_take_the_envelope() {
+ let original = record_with(Some(b""), Some(b"v"), &[]);
+ let message = to_iggy(&original).unwrap();
+ assert!(is_enveloped(&message));
+ let back = from_iggy(&message, 0).unwrap();
+ assert_eq!(back.key.as_deref(), Some(&[][..]), "empty, not null");
+ assert_eq!(back.value.as_deref(), Some(&b"v"[..]));
+ }
+
+ #[test]
+ fn given_an_oversized_key_when_stored_should_take_the_envelope() {
+ let key = vec![b'x'; MAX_FIELD + 1];
+ let message = to_iggy(&record_with(Some(&key), Some(b"v"),
&[])).unwrap();
+ assert!(is_enveloped(&message));
+ assert_eq!(
+ from_iggy(&message, 0).unwrap().key.as_deref(),
+ Some(&key[..])
+ );
+ }
+
+ #[test]
+ fn given_a_null_header_value_when_stored_should_take_the_envelope() {
+ let message = to_iggy(&record_with(Some(b"k"), Some(b"v"), &[("flag",
None)])).unwrap();
+ assert!(is_enveloped(&message));
+ assert_eq!(
+ from_iggy(&message, 0)
+ .unwrap()
+ .headers
+ .get(&StrBytes::from_static_str("flag")),
+ Some(&None),
+ "a null header value survives the envelope as null"
+ );
+ }
+
+ #[test]
+ fn given_an_oversized_header_name_when_stored_should_take_the_envelope() {
+ let name = "n".repeat(MAX_FIELD - HEADER_PREFIX.len() + 1);
+ let message =
+ to_iggy(&record_with(Some(b"k"), Some(b"v"), &[(&name,
Some(b"v"))])).unwrap();
+ assert!(is_enveloped(&message));
+ assert!(
+ from_iggy(&message, 0)
+ .unwrap()
+ .headers
+ .contains_key(&StrBytes::from_string(name))
+ );
+ }
+
+ #[test]
+ fn
given_an_iggy_written_message_when_encoded_should_have_a_null_key_and_its_own_headers()
{
+ let message = message_with(b"{}", &[("source", b"connector")]);
+
+ let record = from_iggy(&message, 3).unwrap();
+ assert_eq!(record.key, None);
+ assert_eq!(record.value.as_deref(), Some(&b"{}"[..]));
+ assert_eq!(
+ record.headers.get(&StrBytes::from_static_str("source")),
+ Some(&Some(Bytes::from_static(b"connector")))
+ );
+ }
+
+ #[test]
+ fn
given_an_iggy_message_with_a_reserved_header_when_read_should_keep_every_header()
{
+ // No `kafka.v`, so nothing here was written by the gateway and the
reserved namespace
+ // carries no meaning. The payload is the value and `own` must not
disappear.
+ let message = message_with(
+ b"payload",
+ &[(VALUE_MARKER_HEADER, MARKER_NULL), ("own", b"1")],
+ );
+
+ let record = from_iggy(&message, 0).unwrap();
+ assert_eq!(
+ record.value.as_deref(),
+ Some(&b"payload"[..]),
+ "not a tombstone"
+ );
+ assert_eq!(
+ record.headers.get(&StrBytes::from_static_str("own")),
+ Some(&Some(Bytes::from_static(b"1")))
+ );
+ assert!(
+ record
+ .headers
+ .contains_key(&StrBytes::from_static_str(VALUE_MARKER_HEADER)),
+ "a reserved name an Iggy client chose passes through under that
name"
+ );
+ }
+
+ #[test]
+ fn
given_a_non_string_key_kind_when_read_should_use_the_bytes_as_the_name() {
+ let mut headers = BTreeMap::new();
+ headers.insert(
+ HeaderKey::from_raw(HeaderKind::Raw, b"trace").unwrap(),
+ header_value(b"abc"),
+ );
+ let message = IggyMessage::builder()
+ .payload(Bytes::from_static(b"v"))
+ .user_headers(headers)
+ .build()
+ .unwrap();
+
+ assert_eq!(
+ from_iggy(&message, 0)
+ .unwrap()
+ .headers
+ .get(&StrBytes::from_static_str("trace")),
+ Some(&Some(Bytes::from_static(b"abc"))),
+ "other SDKs hand out Raw key constructors, and the bytes are a
valid Kafka name"
+ );
+ }
+
+ #[test]
+ fn given_unreadable_user_headers_when_read_should_fail() {
+ let mut message = to_iggy(&record_with(Some(b"k"), Some(b"v"),
&[])).unwrap();
+ message.user_headers = Some(Bytes::from_static(b"not a header block"));
+
+ assert!(
+ matches!(
+ from_iggy(&message, 0),
+ Err(RecordCodecError::UserHeadersUnreadable(_))
+ ),
+ "a block Iggy cannot parse must not read as a message with no
headers"
+ );
+ }
+
+ #[test]
+ fn given_an_unknown_mapping_version_when_read_should_fail() {
+ let message = message_with(b"v", &[(VERSION_HEADER, &[MAPPING_VERSION
+ 1])]);
+ assert!(matches!(
+ from_iggy(&message, 0),
+ Err(RecordCodecError::MappingVersion(2))
+ ));
+ }
+
+ #[test]
+ fn
given_an_unknown_value_marker_on_a_gateway_message_when_read_should_fail() {
+ let message = gateway_message(b"v", &[(VALUE_MARKER_HEADER,
b"neither")]);
+ assert!(
+ matches!(
+ from_iggy(&message, 0),
+ Err(RecordCodecError::ValueMarker(_))
+ ),
+ "a marker this build does not write cannot stand for null or empty"
+ );
+ }
+
+ #[test]
+ fn
given_an_unknown_timestamp_marker_on_a_gateway_message_when_read_should_fail() {
+ let message = gateway_message(b"v", &[(TIMESTAMP_MARKER_HEADER,
b"later")]);
+ assert!(matches!(
+ from_iggy(&message, 0),
+ Err(RecordCodecError::TimestampMarker(_))
+ ));
+ }
+
+ #[test]
+ fn given_a_create_time_when_round_tripped_should_come_back_unchanged() {
+ let message = to_iggy(&record_at(CREATE_TIME)).unwrap();
+ assert_eq!(
+ message.header.origin_timestamp,
+ CREATE_TIME.cast_unsigned() * 1000
+ );
+ assert_eq!(
+ from_iggy(&message, 0).unwrap().timestamp,
+ CREATE_TIME,
+ "a record that arrived through Produce survives the round trip
exactly"
+ );
+ }
+
+ #[test]
+ fn given_no_timestamp_when_stored_should_store_zero() {
+ assert_eq!(timestamp_in(NO_TIMESTAMP).unwrap(), 0);
+ }
+
+ #[test]
+ fn given_an_out_of_range_timestamp_when_stored_should_fail() {
+ assert!(matches!(
+ timestamp_in(i64::MAX),
+ Err(RecordCodecError::TimestampOutOfRange(_))
+ ));
+ }
+
+ #[test]
+ fn given_an_epoch_timestamp_when_round_tripped_should_stay_at_the_epoch() {
+ let mut message = to_iggy(&record_at(EPOCH_TIMESTAMP)).unwrap();
+ message.header.timestamp = 5_000_000;
+ assert_eq!(from_iggy(&message, 0).unwrap().timestamp, EPOCH_TIMESTAMP);
+ }
+
+ #[test]
+ fn
given_an_epoch_timestamp_in_an_envelope_when_read_should_stay_at_the_epoch() {
+ let original = record(
+ Some(Bytes::new()),
+ Some(Bytes::from_static(b"v")),
+ IndexMap::new(),
+ 0,
+ EPOCH_TIMESTAMP,
+ );
+ let mut message = to_iggy(&original).unwrap();
+ assert!(is_enveloped(&message));
+ message.header.timestamp = 5_000_000;
+ assert_eq!(from_iggy(&message, 0).unwrap().timestamp, EPOCH_TIMESTAMP);
+ }
+
+ #[test]
+ fn given_no_timestamp_when_read_should_use_the_server_timestamp() {
+ let mut message = to_iggy(&record_at(NO_TIMESTAMP)).unwrap();
+ message.header.timestamp = 5_000_000;
+ assert_eq!(from_iggy(&message, 0).unwrap().timestamp, 5_000);
+ }
+
+ #[test]
+ fn given_a_truncated_envelope_when_decoded_should_fail() {
+ let message = to_iggy(&record_with(Some(b""), Some(b"v"),
&[])).unwrap();
+ assert!(matches!(
+ decode_envelope(&[ENVELOPE_VERSION], &message.payload.slice(0..3)),
+ Err(RecordCodecError::EnvelopeTruncated { .. })
+ ));
+ }
+
+ #[test]
+ fn given_more_headers_than_the_payload_holds_when_decoded_should_fail() {
+ // Thirteen bytes claiming four billion headers. Reserving for them is
the whole risk.
+ assert!(matches!(
+ decode_envelope(&[ENVELOPE_VERSION], &envelope_bytes(u32::MAX,
b"")),
+ Err(RecordCodecError::EnvelopeTruncated { .. })
+ ));
+ }
+
+ #[test]
+ fn given_bytes_after_the_last_header_when_decoded_should_fail() {
+ assert!(matches!(
+ decode_envelope(&[ENVELOPE_VERSION], &envelope_bytes(0, b"junk")),
+ Err(RecordCodecError::EnvelopeTrailingBytes(4))
+ ));
+ }
+
+ #[test]
+ fn given_a_gateway_envelope_that_does_not_parse_when_read_should_fail() {
+ let message = gateway_message(
+ b"not an envelope",
+ &[(ENVELOPE_HEADER, &[ENVELOPE_VERSION])],
+ );
+ assert!(
+ from_iggy(&message, 0).is_err(),
+ "the gateway writes only envelopes that parse, so this one is
corrupt"
+ );
+ }
+
+ #[test]
+ fn
given_an_envelope_header_without_a_version_header_when_read_should_read_it_as_iggy()
{
+ let message = message_with(
+ b"not an envelope",
+ &[(ENVELOPE_HEADER, &[ENVELOPE_VERSION])],
+ );
+
+ let record = from_iggy(&message, 0).unwrap();
+ assert_eq!(record.key, None, "no Kafka producer wrote this");
+ assert_eq!(record.value.as_deref(), Some(&b"not an envelope"[..]));
+ }
+
+ #[test]
+ fn
given_a_value_filling_the_payload_when_the_envelope_is_needed_should_fail() {
+ // An empty key is the cheapest field Iggy cannot hold, so this record
has to take the
+ // envelope, and the envelope has to carry the key alongside a value
already at the cap.
+ let oversized = record(
+ Some(Bytes::new()),
+ Some(Bytes::from(vec![b'x'; MAX_PAYLOAD_SIZE as usize])),
+ IndexMap::new(),
+ 0,
+ CREATE_TIME,
+ );
+ assert!(matches!(
+ to_iggy(&oversized),
+ Err(RecordCodecError::EnvelopeTooLarge { size })
+ if size == MAX_PAYLOAD_SIZE as usize + ENVELOPE_OVERHEAD
+ ));
+ }
+
+ fn encode_with(records: &[Record], compression: Compression) -> Bytes {
+ let mut buf = BytesMut::new();
+ let options = RecordEncodeOptions {
+ version: BATCH_VERSION,
+ compression,
+ };
+ RecordBatchEncoder::encode(&mut buf, records, &options).unwrap();
+ buf.freeze()
+ }
+
+ fn record_at_offset(offset: i64, value: &[u8]) -> Record {
+ record(
+ Some(Bytes::from_static(b"k")),
+ Some(Bytes::copy_from_slice(value)),
+ IndexMap::new(),
+ offset,
+ CREATE_TIME,
+ )
+ }
+
+ fn batch_count(batch: &Bytes) -> usize {
+ RecordBatchDecoder::decode_batch_info(&mut batch.clone())
+ .unwrap()
+ .len()
+ }
+
+ /// Rewrites a batch header field and repairs the CRC, so the decoder
reaches the field.
+ ///
+ /// The CRC covers every byte after it, and it is checked before the
record count is read, so
+ /// a patched count without a repaired CRC only ever tests CRC
verification.
+ fn patch_header(batch: &Bytes, offset: usize, value: &[u8]) -> Bytes {
+ let mut bytes = batch.to_vec();
+ bytes[offset..offset + value.len()].copy_from_slice(value);
+ let crc = crc32c::crc32c(&bytes[21..]);
+ bytes[17..21].copy_from_slice(&crc.to_be_bytes());
+ Bytes::from(bytes)
+ }
+
+ #[test]
+ fn
given_an_uncompressed_batch_when_round_tripped_should_keep_every_record() {
+ let mut records = vec![record_at_offset(0, b"1"), record_at_offset(1,
b"2")];
+ let mut encoded = encode_batch(&mut records).unwrap();
+ let budget = DecompressionBudget::new(1024);
+ let decoded = decode_batches(&mut encoded, &budget).unwrap();
+ assert_eq!(decoded.len(), 2);
+ assert_eq!(decoded[1].value.as_deref(), Some(&b"2"[..]));
+ }
+
+ #[test]
+ fn given_records_at_distinct_offsets_when_encoded_should_make_one_batch() {
+ let mut records = (0..4)
+ .map(|offset| record_at_offset(offset, b"v"))
+ .collect::<Vec<_>>();
+ let encoded = encode_batch(&mut records).unwrap();
+
+ assert_eq!(
+ batch_count(&encoded),
+ 1,
+ "a batch header per record costs 61 bytes of framing on every
Fetch"
+ );
+ assert_eq!(
+ records[0].sequence, NO_SEQUENCE,
+ "the encoded base sequence follows the first record's"
+ );
+ }
+
+ #[test]
+ fn given_no_records_when_encoded_should_write_nothing() {
+ assert!(encode_batch(&mut []).unwrap().is_empty());
+ }
+
+ #[test]
+ fn given_two_batches_in_one_blob_when_decoded_should_drain_both() {
+ let mut blob = BytesMut::new();
+ blob.extend_from_slice(&encode_batch(&mut [record_at_offset(0,
b"1")]).unwrap());
+ blob.extend_from_slice(&encode_batch(&mut [record_at_offset(1,
b"2")]).unwrap());
+
+ let budget = DecompressionBudget::new(1024);
+ let decoded = decode_batches(&mut blob.freeze(), &budget).unwrap();
+ assert_eq!(decoded.len(), 2, "a partition blob can hold many batches");
+ }
+
+ #[test]
+ fn given_a_gzip_batch_when_decoded_should_read_it() {
+ let records = vec![record_at_offset(0, b"compressed")];
+ let mut encoded = encode_with(&records, Compression::Gzip);
+ let budget = DecompressionBudget::new(1024);
+ let decoded = decode_batches(&mut encoded, &budget).unwrap();
+ assert_eq!(decoded[0].value.as_deref(), Some(&b"compressed"[..]));
+ }
+
+ #[test]
+ fn given_a_snappy_batch_when_decoded_should_read_it() {
+ let records = vec![record_at_offset(0, b"compressed")];
+ let mut encoded = encode_with(&records, Compression::Snappy);
+ let budget = DecompressionBudget::new(1024);
+ let decoded = decode_batches(&mut encoded, &budget).unwrap();
+ assert_eq!(decoded[0].value.as_deref(), Some(&b"compressed"[..]));
+ }
+
+ #[test]
+ fn given_an_lz4_batch_when_decoded_should_read_it() {
+ let records = vec![record_at_offset(0, b"compressed")];
+ let mut encoded = encode_with(&records, Compression::Lz4);
+ let budget = DecompressionBudget::new(1024);
+ let decoded = decode_batches(&mut encoded, &budget).unwrap();
+ assert_eq!(decoded[0].value.as_deref(), Some(&b"compressed"[..]));
+ }
+
+ #[test]
+ fn given_a_zstd_batch_when_decoded_should_read_it() {
+ let records = vec![record_at_offset(0, b"compressed")];
+ let mut encoded = encode_with(&records, Compression::Zstd);
+ let budget = DecompressionBudget::new(1024);
+ let decoded = decode_batches(&mut encoded, &budget).unwrap();
+ assert_eq!(decoded[0].value.as_deref(), Some(&b"compressed"[..]));
+ }
+
+ #[test]
+ fn given_a_budget_smaller_than_the_batch_when_decoded_should_reject() {
+ let records = vec![record_at_offset(0, &[b'x'; 512])];
+ let mut encoded = encode_with(&records, Compression::Gzip);
+ let budget = DecompressionBudget::new(8);
+ assert!(matches!(
+ decode_batches(&mut encoded, &budget),
+ Err(RecordCodecError::BudgetExceeded { .. })
+ ));
+ }
+
+ #[test]
+ fn given_two_batches_when_the_second_passes_the_budget_should_reject() {
+ let big = record_at_offset(0, &[b'x'; 256]);
+ let mut blob = BytesMut::new();
+ blob.extend_from_slice(&encode_batch(&mut [big.clone()]).unwrap());
+ blob.extend_from_slice(&encode_batch(&mut [big]).unwrap());
+
+ // Enough for one batch, not for both: the budget is per request, not
per batch.
+ let budget = DecompressionBudget::new(400);
+ assert!(matches!(
+ decode_batches(&mut blob.freeze(), &budget),
+ Err(RecordCodecError::BudgetExceeded { .. })
+ ));
+ }
+
+ #[test]
+ fn given_a_compression_bomb_when_decoded_should_stop_before_it_is_whole() {
+ let bomb = vec![0u8; 4 * 1024 * 1024];
+ let mut compressed = Bytes::from(encode_gzip(&bomb));
+ let budget = DecompressionBudget::new(1024);
+
+ let Err(error) = decompress(&mut compressed, Compression::Gzip,
&budget) else {
+ panic!("a 4 MB output against a 1 KB budget has to be refused");
+ };
+ drop(error);
+ let Some(RecordCodecError::BudgetExceeded {
+ produced,
+ remaining,
+ }) = budget.reason.take()
+ else {
+ panic!("the budget is what refused it");
+ };
+ assert_eq!(remaining, 1024);
+ assert!(
+ produced < bomb.len(),
+ "the write that passed the budget was refused, not charged
afterwards: \
+ {produced} of {} bytes",
+ bomb.len()
+ );
+ }
+
+ #[test]
+ fn
given_a_snappy_block_declaring_more_than_the_budget_when_decoded_should_reject()
{
+ // A raw snappy stream whose leading varint claims u32::MAX bytes of
output. `snap` reads
+ // that length without allocating, so the declared size is checkable
before the decode.
+ let mut compressed = Bytes::from_static(&[0xff, 0xff, 0xff, 0xff,
0x0f, 0x00]);
+ let budget = DecompressionBudget::new(1024);
+ assert!(decompress(&mut compressed, Compression::Snappy,
&budget).is_err());
+ assert!(matches!(
+ budget.reason.take(),
+ Some(RecordCodecError::BudgetExceeded { produced, .. }) if
produced == u32::MAX as usize
+ ));
+ }
+
+ #[test]
+ fn given_a_record_count_past_the_frame_when_decoded_should_reject() {
+ let batch = encode_batch(&mut [record_at_offset(0, b"v")]).unwrap();
+ let mut patched = patch_header(&batch, 57, &i32::MAX.to_be_bytes());
+ let budget = DecompressionBudget::new(1024);
+
+ assert!(
+ matches!(
+ decode_batches(&mut patched, &budget),
+ Err(RecordCodecError::RecordCountTooLarge { .. })
+ ),
+ "the decoder reserves from this count before it reads a record"
+ );
+ }
+
+ #[test]
+ fn given_counts_that_only_pass_one_at_a_time_when_decoded_should_reject() {
+ // One large batch raises what the blob can hold, and every small
batch then declares a
+ // count that clears the bound on its own. The reserves land in one
`Vec` all the same.
+ let big = encode_batch(&mut [record_at_offset(0, &[b'x';
2048])]).unwrap();
+ let small = patch_header(
+ &encode_batch(&mut [record_at_offset(0, b"v")]).unwrap(),
+ 57,
+ &200i32.to_be_bytes(),
+ );
+
+ let mut blob = BytesMut::new();
+ blob.extend_from_slice(&big);
+ for _ in 0..8 {
+ blob.extend_from_slice(&small);
+ }
+
+ let budget = DecompressionBudget::new(0);
+ assert!(matches!(
+ decode_batches(&mut blob.freeze(), &budget),
+ Err(RecordCodecError::RecordCountTooLarge { .. })
+ ));
+ }
+
+ /// A one-record v2 batch whose record body declares `headers` headers and
carries none.
+ ///
+ /// Built by hand because no encoder writes that, and it is the shape that
reaches
+ /// `IndexMap::with_capacity` inside kafka-protocol.
+ fn batch_declaring_headers(headers: i32) -> Bytes {
+ fn put_varint(buf: &mut BytesMut, value: i32) {
+ let mut zigzag = ((value << 1) ^ (value >> 31)).cast_unsigned();
+ while zigzag >= 0x80 {
+ buf.put_u8(u8::try_from(zigzag & 0x7f).unwrap() | 0x80);
+ zigzag >>= 7;
+ }
+ buf.put_u8(u8::try_from(zigzag).unwrap());
+ }
+
+ let mut record = BytesMut::new();
+ record.put_u8(0); // attributes
+ put_varint(&mut record, 0); // timestamp delta
+ put_varint(&mut record, 0); // offset delta
+ put_varint(&mut record, -1); // null key
+ put_varint(&mut record, -1); // null value
+ put_varint(&mut record, headers);
+
+ let mut records = BytesMut::new();
+ put_varint(&mut records, i32::try_from(record.len()).unwrap());
+ records.extend_from_slice(&record);
+
+ // Everything from the attributes field on, which is what the CRC
covers.
+ let mut body = BytesMut::new();
+ body.put_i16(0); // attributes: no compression, CreateTime
+ body.put_i32(0); // last offset delta
+ body.put_i64(CREATE_TIME); // first timestamp
+ body.put_i64(CREATE_TIME); // max timestamp
+ body.put_i64(NO_PRODUCER_ID);
+ body.put_i16(NO_PRODUCER_EPOCH);
+ body.put_i32(NO_SEQUENCE);
+ body.put_i32(1); // record count
+ body.extend_from_slice(&records);
+
+ let mut batch = BytesMut::new();
+ batch.put_i64(0); // base offset
+ batch.put_i32(i32::try_from(body.len() + 9).unwrap()); // batch
length, from leader epoch on
+ batch.put_i32(NO_PARTITION_LEADER_EPOCH);
+ batch.put_i8(BATCH_VERSION);
+ batch.put_u32(crc32c::crc32c(&body));
+ batch.extend_from_slice(&body);
+ batch.freeze()
+ }
+
+ #[test]
+ fn given_a_header_count_past_the_record_when_decoded_should_reject() {
+ // The batch header says one record, so the record count bound passes.
The count that
+ // matters is the one inside the record, and no batch header reports
it.
+ let mut batch = batch_declaring_headers(i32::MAX);
+ let budget = DecompressionBudget::new(8 * 1024 * 1024);
+
+ assert!(
+ matches!(
+ decode_batches(&mut batch, &budget),
+ Err(RecordCodecError::HeaderCountTooLarge { count, .. }) if
count == i32::MAX
+ ),
+ "this reserve is resident memory, not address space"
+ );
+ }
+
+ #[test]
+ fn given_a_hand_built_batch_when_its_header_count_fits_should_decode() {
+ // The same builder with a count the record can hold, so the scan
cannot be passing the
+ // test above by rejecting every hand-built batch.
+ let mut batch = batch_declaring_headers(0);
+ let budget = DecompressionBudget::new(1024);
+ assert_eq!(decode_batches(&mut batch, &budget).unwrap().len(), 1);
+ }
+
+ #[test]
+ fn given_records_carrying_headers_when_round_tripped_should_keep_them() {
+ // Exercises the header framing the scan walks, on a batch an encoder
wrote.
+ let mut with_headers = vec![record_with(
+ Some(b"k"),
+ Some(b"v"),
+ &[("trace", Some(b"abc"))],
+ )];
+ let mut encoded = encode_batch(&mut with_headers).unwrap();
+ let budget = DecompressionBudget::new(1024);
+
+ let decoded = decode_batches(&mut encoded, &budget).unwrap();
+ assert_eq!(
+ decoded[0].headers.get(&StrBytes::from_static_str("trace")),
+ Some(&Some(Bytes::from_static(b"abc")))
+ );
+ }
+
+ #[test]
+ fn
given_an_uncompressed_batch_when_preflighting_should_not_grant_it_the_budget() {
+ // One record's worth of bytes, declaring far more records than it
holds. The budget is
+ // the documented 8 MiB, which an uncompressed batch has no claim on.
+ let batch = encode_batch(&mut [record_at_offset(0, b"v")]).unwrap();
+ let mut patched = patch_header(&batch, 57, &100_000i32.to_be_bytes());
+ let budget = DecompressionBudget::new(8 * 1024 * 1024);
+
+ assert!(matches!(
+ decode_batches(&mut patched, &budget),
+ Err(RecordCodecError::RecordCountTooLarge { limit, .. }) if limit
== batch.len()
+ ));
+ }
+
+ #[test]
+ fn
given_two_key_kinds_naming_one_header_on_a_gateway_message_when_read_should_fail()
{
+ let mut headers = BTreeMap::new();
+ headers.insert(header_key(VERSION_HEADER),
header_value(&[MAPPING_VERSION]));
+ headers.insert(header_key("kafka.h.trace"), header_value(b"string"));
+ headers.insert(
+ HeaderKey::from_raw(HeaderKind::Raw, b"kafka.h.trace").unwrap(),
+ header_value(b"raw"),
+ );
+ let message = IggyMessage::builder()
+ .payload(Bytes::from_static(b"v"))
+ .user_headers(headers)
+ .build()
+ .unwrap();
+
+ assert!(
+ matches!(
+ from_iggy(&message, 0),
+ Err(RecordCodecError::HeaderNameCollision(_))
+ ),
+ "to_iggy writes one key kind, so a collision means the message is
not what it claims"
+ );
+ }
+
+ #[test]
+ fn
given_two_key_kinds_naming_one_header_on_an_iggy_message_when_read_should_keep_one()
{
+ let mut headers = BTreeMap::new();
+ headers.insert(header_key("trace"), header_value(b"string"));
+ headers.insert(
+ HeaderKey::from_raw(HeaderKind::Raw, b"trace").unwrap(),
+ header_value(b"raw"),
+ );
+ let message = IggyMessage::builder()
+ .payload(Bytes::from_static(b"v"))
+ .user_headers(headers)
+ .build()
+ .unwrap();
+
+ let record = from_iggy(&message, 0).unwrap();
+ assert_eq!(
+ record.headers.len(),
+ 1,
+ "a Record keys headers in an IndexMap, so the pair cannot both
survive"
+ );
+ assert!(
+ record
+ .headers
+ .contains_key(&StrBytes::from_static_str("trace")),
+ "refusing the message instead would stall the partition for every
Kafka consumer"
+ );
+ }
+
+ #[test]
+ fn given_a_transactional_batch_when_decoded_should_reject() {
+ let mut transactional = record_at_offset(0, b"v");
+ transactional.transactional = true;
+ let mut encoded = encode_with(&[transactional], Compression::None);
+ let budget = DecompressionBudget::new(1024);
+
+ assert!(matches!(
+ decode_batches(&mut encoded, &budget),
+ Err(RecordCodecError::UnsupportedBatch("transactional"))
+ ));
+ }
+
+ #[test]
+ fn given_a_control_batch_when_decoded_should_reject() {
+ let mut control = record_at_offset(0, b"v");
+ control.control = true;
+ let mut encoded = encode_with(&[control], Compression::None);
+ let budget = DecompressionBudget::new(1024);
+
+ assert!(
+ matches!(
+ decode_batches(&mut encoded, &budget),
+ Err(RecordCodecError::UnsupportedBatch("control"))
+ ),
+ "consumers filter control records by a flag no stored message can
carry"
+ );
+ }
+
+ #[test]
+ fn given_a_truncated_batch_when_decoded_should_reject() {
+ let batch = encode_batch(&mut [record_at_offset(0,
b"value")]).unwrap();
+ let mut truncated = batch.slice(0..batch.len() - 4);
+ let budget = DecompressionBudget::new(1024);
+ assert!(decode_batches(&mut truncated, &budget).is_err());
+ }
+
+ #[test]
+ fn given_a_corrupt_batch_crc_when_decoded_should_reject() {
+ let batch = encode_batch(&mut [record_at_offset(0,
b"value")]).unwrap();
+ let mut bytes = batch.to_vec();
+ let last = bytes.len() - 1;
+ bytes[last] ^= 0xff;
+ let mut corrupt = Bytes::from(bytes);
+
+ let budget = DecompressionBudget::new(1024);
+ assert!(matches!(
+ decode_batches(&mut corrupt, &budget),
+ Err(RecordCodecError::Batch(_))
+ ));
+ }
+
+ #[test]
+ fn
given_an_earlier_overrun_when_a_later_batch_fails_should_not_reuse_the_budget_reason()
{
+ // A failed charge deducts nothing, so the budget still has room for
the batch below.
+ let budget = DecompressionBudget::new(64);
+ let mut bomb = encode_with(&[record_at_offset(0, &[b'x'; 512])],
Compression::Gzip);
+ assert!(matches!(
+ decode_batches(&mut bomb, &budget),
+ Err(RecordCodecError::BudgetExceeded { .. })
+ ));
+
+ // One record more than the batch holds. The count clears the
preflight bound, so the
+ // failure comes out of the record decoder, which is the path that
reads the overflow.
+ let batch = encode_batch(&mut [record_at_offset(0,
b"value")]).unwrap();
+ let mut short = patch_header(&batch, 57, &2i32.to_be_bytes());
+
+ assert!(
+ matches!(
+ decode_batches(&mut short, &budget),
+ Err(RecordCodecError::Batch(_))
+ ),
+ "a stale overflow reports a malformed batch as MESSAGE_TOO_LARGE"
+ );
+ }
+
+ fn encode_gzip(body: &[u8]) -> Vec<u8> {
+ let mut encoder = flate2::write::GzEncoder::new(Vec::new(),
flate2::Compression::default());
+ encoder.write_all(body).unwrap();
+ encoder.finish().unwrap()
+ }
+}
