ryerraguntla commented on code in PR #4229: URL: https://github.com/apache/iggy/pull/4229#discussion_r4058481787
########## 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: any kafka.envelope forces envelope path. Iggy garbage → Fetch fail or :264 OOM. Doc reserve only. -- 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]
