krishvishal commented on code in PR #4229: URL: https://github.com/apache/iggy/pull/4229#discussion_r4062266216
########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,723 @@ +// 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 batch framing, no Iggy calls, no handler wiring. + +use std::cell::Cell; +use std::collections::BTreeMap; + +use bytes::{Buf, BufMut, Bytes, BytesMut}; +use iggy::prelude::{HeaderKey, HeaderValue, IggyError, IggyMessage}; +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 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"; +/// 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 format version. +pub const ENVELOPE_HEADER: &str = "kafka.envelope"; +/// Envelope format 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; +/// 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 FLAG_KEY: u8 = 0b01; +const FLAG_VALUE: u8 = 0b10; + +/// Record batch version this gateway writes. v2 is the only shape `kafka_protocol` encodes. +const BATCH_VERSION: i8 = 2; + +#[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("envelope is truncated: needed {needed} bytes, {remaining} remain")] + EnvelopeTruncated { needed: usize, remaining: 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("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, or when Iggy rejects the message for a +/// reason the envelope does not fix, such as a payload over `MAX_PAYLOAD_SIZE`. +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 = BTreeMap::new(); + 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 with no `kafka.` headers was written by an Iggy client, not through this gateway. +/// It gets a null key and its own user headers. +/// +/// # Errors +/// +/// Returns an error when the message carries a malformed envelope. +pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> { + let stored = message.user_headers_map()?.unwrap_or_default(); + let envelope = stored.get(&header_key(ENVELOPE_HEADER)); + let (key, value, headers) = if let Some(version) = envelope { + decode_envelope(version.value().as_ref(), &message.payload)? + } else { + native_fields(&stored, message) + }; + Ok(record(key, value, headers, offset, timestamp_out(message))) +} + +/// 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. +fn timestamp_out(message: &IggyMessage) -> i64 { + let micros = if message.header.origin_timestamp == 0 { Review Comment: Fixed ########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,723 @@ +// 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 batch framing, no Iggy calls, no handler wiring. + +use std::cell::Cell; +use std::collections::BTreeMap; + +use bytes::{Buf, BufMut, Bytes, BytesMut}; +use iggy::prelude::{HeaderKey, HeaderValue, IggyError, IggyMessage}; +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 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"; +/// 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 format version. +pub const ENVELOPE_HEADER: &str = "kafka.envelope"; +/// Envelope format 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; +/// 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 FLAG_KEY: u8 = 0b01; +const FLAG_VALUE: u8 = 0b10; + +/// Record batch version this gateway writes. v2 is the only shape `kafka_protocol` encodes. +const BATCH_VERSION: i8 = 2; + +#[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("envelope is truncated: needed {needed} bytes, {remaining} remain")] + EnvelopeTruncated { needed: usize, remaining: 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("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, or when Iggy rejects the message for a +/// reason the envelope does not fix, such as a payload over `MAX_PAYLOAD_SIZE`. +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 = BTreeMap::new(); + 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 with no `kafka.` headers was written by an Iggy client, not through this gateway. +/// It gets a null key and its own user headers. +/// +/// # Errors +/// +/// Returns an error when the message carries a malformed envelope. +pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> { + let stored = message.user_headers_map()?.unwrap_or_default(); + let envelope = stored.get(&header_key(ENVELOPE_HEADER)); + let (key, value, headers) = if let Some(version) = envelope { + decode_envelope(version.value().as_ref(), &message.payload)? + } else { + native_fields(&stored, message) + }; + Ok(record(key, value, headers, offset, timestamp_out(message))) +} + +/// 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. +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), + } +} + +/// `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> { + let mut headers = BTreeMap::new(); + headers.insert( + header_key(ENVELOPE_HEADER), + header_value(&[ENVELOPE_VERSION]), + ); + build(encode_envelope(record), headers, record.timestamp)? + .ok_or(IggyError::TooBigUserHeaders) + .map_err(Into::into) +} + +/// 13 bytes of fixed overhead plus 9 per header, little-endian throughout. +fn encode_envelope(record: &Record) -> 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::new(); + 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 EnvelopeFields = ( + Option<Bytes>, + Option<Bytes>, + IndexMap<StrBytes, Option<Bytes>>, +); + +fn decode_envelope(version: &[u8], payload: &Bytes) -> Result<EnvelopeFields> { + 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()); + + let mut headers = IndexMap::with_capacity(count as usize); + 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)); + } + + Ok(( + (flags & FLAG_KEY != 0).then_some(key), + (flags & FLAG_VALUE != 0).then_some(value), + headers, + )) +} + +fn native_fields( + stored: &BTreeMap<HeaderKey, HeaderValue>, + message: &IggyMessage, +) -> EnvelopeFields { + let gateway_written = stored + .keys() + .any(|key| key.as_str().is_ok_and(|key| key.starts_with("kafka."))); + let key = stored.get(&header_key(KEY_HEADER)).map(HeaderValue::value); + let value = match stored + .get(&header_key(VALUE_MARKER_HEADER)) + .map(HeaderValue::value) + { + Some(marker) if marker.as_ref() == MARKER_NULL => None, + Some(_) => Some(Bytes::new()), Review Comment: Fixed ########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,723 @@ +// 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 batch framing, no Iggy calls, no handler wiring. + +use std::cell::Cell; +use std::collections::BTreeMap; + +use bytes::{Buf, BufMut, Bytes, BytesMut}; +use iggy::prelude::{HeaderKey, HeaderValue, IggyError, IggyMessage}; +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 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"; +/// 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 format version. +pub const ENVELOPE_HEADER: &str = "kafka.envelope"; +/// Envelope format 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; +/// 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 FLAG_KEY: u8 = 0b01; +const FLAG_VALUE: u8 = 0b10; + +/// Record batch version this gateway writes. v2 is the only shape `kafka_protocol` encodes. +const BATCH_VERSION: i8 = 2; + +#[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("envelope is truncated: needed {needed} bytes, {remaining} remain")] + EnvelopeTruncated { needed: usize, remaining: 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("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, or when Iggy rejects the message for a +/// reason the envelope does not fix, such as a payload over `MAX_PAYLOAD_SIZE`. +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 = BTreeMap::new(); + 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 with no `kafka.` headers was written by an Iggy client, not through this gateway. +/// It gets a null key and its own user headers. +/// +/// # Errors +/// +/// Returns an error when the message carries a malformed envelope. +pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> { + let stored = message.user_headers_map()?.unwrap_or_default(); + let envelope = stored.get(&header_key(ENVELOPE_HEADER)); + let (key, value, headers) = if let Some(version) = envelope { + decode_envelope(version.value().as_ref(), &message.payload)? + } else { + native_fields(&stored, message) + }; + Ok(record(key, value, headers, offset, timestamp_out(message))) +} + +/// 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. +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), + } +} + +/// `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> { + let mut headers = BTreeMap::new(); + headers.insert( + header_key(ENVELOPE_HEADER), + header_value(&[ENVELOPE_VERSION]), + ); + build(encode_envelope(record), headers, record.timestamp)? Review Comment: Fixed ########## gateways/kafka/src/records.rs: ########## @@ -0,0 +1,723 @@ +// 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 batch framing, no Iggy calls, no handler wiring. + +use std::cell::Cell; +use std::collections::BTreeMap; + +use bytes::{Buf, BufMut, Bytes, BytesMut}; +use iggy::prelude::{HeaderKey, HeaderValue, IggyError, IggyMessage}; +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 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"; +/// 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 format version. +pub const ENVELOPE_HEADER: &str = "kafka.envelope"; +/// Envelope format 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; +/// 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 FLAG_KEY: u8 = 0b01; +const FLAG_VALUE: u8 = 0b10; + +/// Record batch version this gateway writes. v2 is the only shape `kafka_protocol` encodes. +const BATCH_VERSION: i8 = 2; + +#[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("envelope is truncated: needed {needed} bytes, {remaining} remain")] + EnvelopeTruncated { needed: usize, remaining: 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("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, or when Iggy rejects the message for a +/// reason the envelope does not fix, such as a payload over `MAX_PAYLOAD_SIZE`. +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 = BTreeMap::new(); + 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 with no `kafka.` headers was written by an Iggy client, not through this gateway. +/// It gets a null key and its own user headers. +/// +/// # Errors +/// +/// Returns an error when the message carries a malformed envelope. +pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> { + let stored = message.user_headers_map()?.unwrap_or_default(); + let envelope = stored.get(&header_key(ENVELOPE_HEADER)); + let (key, value, headers) = if let Some(version) = envelope { Review Comment: Fixed ########## gateways/kafka/docs/BRIDGE_MAPPING.md: ########## @@ -161,9 +165,16 @@ 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 what a request accumulates, not what one batch allocates. `kafka_protocol`'s +decompressors write the whole stream out before they hand it over (`compression/gzip.rs:46` and +its three siblings), so a single batch reaches its full decompressed size in memory and the +budget rejects it one step later. Bounding the peak needs a size-limited reader per codec, which +means owning Kafka's snappy and lz4 framing rather than borrowing it. That is worth doing and it +is not done here. Nothing decompresses today. The record batch stays an opaque `Bytes` on both paths, so the bound Review Comment: Fixed -- 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]
