numinnex commented on code in PR #4229: URL: https://github.com/apache/iggy/pull/4229#discussion_r4062556981
########## gateways/kafka/docs/BRIDGE_MAPPING.md: ########## @@ -148,9 +210,26 @@ 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. + +Produce reads every batch header before it decodes a record. The header's record count is four +bytes a client chooses, and upstream checks it for sign alone. `RecordBatchDecoder` then reserves +from it (`kafka-protocol-0.18.0/src/records.rs:517`), so a 61-byte batch declaring `i32::MAX` +records asks for 377 GB. The records can come to no more than the frame plus what the +decompression budget still allows. A count that needs more is refused before anything decodes. Review Comment: **This states a guarantee the code does not provide.** "A count that needs more is refused before anything decodes" is true of the batch-level `record_count` and false of the per-record `num_headers`, which `preflight` cannot see. The unbounded one is the one that aborts the process. A doc asserting the class is closed is worse than no doc here, because it is what a future reader will check against. Suggest naming which count is bounded, and recording `num_headers` as open until it is. ########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,1518 @@ +// 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; +/// 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("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("{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` was written here, so every marker on it is authoritative and one this +/// build does not recognize is an error rather than a guess. +/// +/// # 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; + }; + headers.insert( + StrBytes::from_string(name.to_string()), + Some(stored_value.value()), + ); + } + 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. +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>, + /// What the charge was when it first tripped. Decompression reports failure through + /// `io::Error` and `anyhow`, which the decoder stringifies, so the typed reason is kept here. + overflow: Cell<Option<(usize, usize)>>, +} + +impl DecompressionBudget { + #[must_use] + pub const fn new(bytes: usize) -> Self { + Self { + remaining: Cell::new(bytes), + overflow: Cell::new(None), + } + } + + fn charge(&self, produced: usize) -> io::Result<()> { + let remaining = self.remaining.get(); + if produced > remaining { + self.overflow.set(Some((produced, remaining))); + return Err(io::Error::other(format!( + "decompressed {produced} bytes with {remaining} left in the budget" + ))); + } + self.remaining.set(remaining - produced); + Ok(()) + } + + /// The typed reason for a decoder error, when the budget is what caused it. + /// + /// Taken rather than read, because the budget outlives one batch. Left in place, the first + /// overrun would reclassify every later error on the same request as a budget overrun. + fn overflow(&self, error: &str) -> RecordCodecError { + self.overflow.take().map_or_else( + || RecordCodecError::Batch(error.to_string()), + |(produced, remaining)| RecordCodecError::BudgetExceeded { + produced, + remaining, + }, + ) + } +} + +/// 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.overflow(&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 frame plus what the budget still +/// allows is the most the records of the whole blob can come to, so the counts are charged against +/// that together rather than one batch at a time: 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 is left is the ratio between a minimal wire record and a decoded `Record`, which one +/// legitimate request pays as well. This bounds the count, not that ratio. +/// +/// 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<()> { Review Comment: **Blocker: `num_headers` is still unbounded, and it is the half that aborts the process.** `decode_batch_info` returns batch headers only, so the per-record header count - a zigzag varint inside a record body - is invisible here by construction. A ~72-byte valid-CRC blob with `record_count=1` and a header-count varint of `i32::MAX` clears this check and reaches `IndexMap::with_capacity` in kafka-protocol. Reproduced: `exit 134`, and unlike `Vec::reserve` this is resident memory, because hashbrown memsets its control array. The bounded path was the address-space one; this is the one the earlier review measured at 4.0 GiB RSS. Suggest pre-scanning record bodies here, or rejecting a `num_headers` the record's own declared size cannot hold (>= 2 bytes per header), plus a byte-literal test. ########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,1518 @@ +// 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; +/// 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("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("{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` was written here, so every marker on it is authoritative and one this Review Comment: **A forged `kafka.v` wedges the partition, and this rustdoc contradicts line 41.** Line 41 says the namespace is "reserved by convention only" and nothing on the server enforces it, so any Iggy-plane writer can set `kafka.v = [2]`. Every `from_iggy` on that message then returns `MappingVersion(2)`, and a Kafka consumer cannot advance past a record the broker refuses to serve: one poison message stalls the partition. The provenance rewrite is a real improvement over the prefix scan, but it replaced a weak guarantee with an asserted one. Suggest stating a skip or quarantine policy for #3536, and softening this claim to match line 41. ########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,1518 @@ +// 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; +/// 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("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("{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` was written here, so every marker on it is authoritative and one this +/// build does not recognize is an error rather than a guess. +/// +/// # 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; + }; + headers.insert( + StrBytes::from_string(name.to_string()), + Some(stored_value.value()), + ); + } + 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. +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>, + /// What the charge was when it first tripped. Decompression reports failure through + /// `io::Error` and `anyhow`, which the decoder stringifies, so the typed reason is kept here. + overflow: Cell<Option<(usize, usize)>>, +} + +impl DecompressionBudget { + #[must_use] + pub const fn new(bytes: usize) -> Self { + Self { + remaining: Cell::new(bytes), + overflow: Cell::new(None), + } + } + + fn charge(&self, produced: usize) -> io::Result<()> { + let remaining = self.remaining.get(); + if produced > remaining { + self.overflow.set(Some((produced, remaining))); + return Err(io::Error::other(format!( + "decompressed {produced} bytes with {remaining} left in the budget" + ))); + } + self.remaining.set(remaining - produced); + Ok(()) + } + + /// The typed reason for a decoder error, when the budget is what caused it. + /// + /// Taken rather than read, because the budget outlives one batch. Left in place, the first + /// overrun would reclassify every later error on the same request as a budget overrun. + fn overflow(&self, error: &str) -> RecordCodecError { + self.overflow.take().map_or_else( + || RecordCodecError::Batch(error.to_string()), + |(produced, remaining)| RecordCodecError::BudgetExceeded { + produced, + remaining, + }, + ) + } +} + +/// 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.overflow(&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 frame plus what the budget still +/// allows is the most the records of the whole blob can come to, so the counts are charged against +/// that together rather than one batch at a time: 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 is left is the ratio between a minimal wire record and a decoded `Record`, which one +/// legitimate request pays as well. This bounds the count, not that ratio. +/// +/// 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 limit = buf.len().saturating_add(budget.remaining.get()); Review Comment: **Uncompressed batches are granted the full decompression budget as headroom.** `buf.len() + budget.remaining` is right for a compressed batch, but an uncompressed one's records cannot exceed its own byte length, so this inflates the ceiling by the whole budget. Measured: a 70-byte batch at the documented 8 MiB budget declares 1,198,382 records and reserves ~210 MB. `Vec::reserve` means address space rather than RSS, so this is not the blocker above - but it is the same amplification preflight exists to close, and `BatchDecodeInfo.compression` is already in hand. Suggest adding the budget term only when the batch is compressed. ########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,1518 @@ +// 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; +/// 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("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("{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` was written here, so every marker on it is authoritative and one this +/// build does not recognize is an error rather than a guess. +/// +/// # 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; + }; + headers.insert( + StrBytes::from_string(name.to_string()), + Some(stored_value.value()), + ); + } + 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. +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> { Review Comment: **The key-kind fix introduces a silent collision.** `HeaderField: Ord` sorts on `(kind, bytes)`, so a String key `a` and a Raw key `a` are two live `BTreeMap` entries. Reading the bytes and letting UTF-8 decide is the right call, but it now maps both to one Kafka header, last-write-wins, with no signal. The old code dropped every non-String key; this drops one of a colliding pair, which is harder to notice. Not covered by the new test. Suggest erroring on collision, or documenting it and adding a test either way. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
