numinnex commented on code in PR #4398: URL: https://github.com/apache/iggy/pull/4398#discussion_r4194786643
########## gateways/kafka/src/protocol/handlers/alter_configs.rs: ########## @@ -0,0 +1,325 @@ +// 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. + +//! `AlterConfigs` (API key 33). + +use bytes::Bytes; +use iggy::prelude::IggyError; +use kafka_protocol::messages::AlterConfigsRequest; +use kafka_protocol::messages::alter_configs_request::AlterConfigsResource; +use kafka_protocol::messages::alter_configs_response::{ + AlterConfigsResourceResponse, AlterConfigsResponse, +}; +use kafka_protocol::protocol::StrBytes; +use tokio::time::Instant; + +use crate::bridge::topic_map::validate_kafka_topic_name; +use crate::bridge::{BridgeError, IggyBridge, StreamTopicCache, TopicLoad}; +use crate::protocol::api::{ + API_KEY_ALTER_CONFIGS, ApiVersionRange, ERROR_INVALID_CONFIG, ERROR_INVALID_REQUEST, + ERROR_INVALID_TOPIC_EXCEPTION, ERROR_NONE, ERROR_NOT_CONTROLLER, ERROR_POLICY_VIOLATION, + ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, + REQUEST_DEADLINE, +}; +use crate::protocol::bounds_guard::validate_alter_configs_shape; +use crate::protocol::handlers::topic_config::{ + IggyTopicKey, ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC, bridge_failure, exceeds_topic_cap, + find_duplicate_names, name_reason, plan_retention_update, static_text, topic_cap_message, +}; +use crate::protocol::handlers::{ + decode_guarded, encode_message, is_supported_version, respond_or_close, + unsupported_version_response, +}; + +pub const RANGE: ApiVersionRange = ApiVersionRange { + api_key: API_KEY_ALTER_CONFIGS, + min_version: 0, + max_version: 2, +}; + +pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { Review Comment: no authz before an API that deletes data: any admitted client can set `retention.ms=1` and the cleaner drops sealed segments irreversibly (no consumer barrier, the gateway has no OffsetCommit). With SASL on, a read-only principal can do it while DescribeAcls tells it ALTER is denied. README.md:9 warns only about CreateTopics. With SASL on, gate on `principal.permissions.manage_topics` (auth.rs:512) and answer 29 per resource. At minimum, add the README warning and a CONFIGS.md authz section that names `retention.ms` as destructive. ########## gateways/kafka/src/protocol/bounds_guard.rs: ########## @@ -1052,6 +1052,110 @@ pub fn validate_sync_group_shape(version: i16, body: &Bytes, max_frame_size: usi /// reaches `parse_plain`, and what a future mechanism would carry into a credential exchange. const MAX_SASL_AUTH_BYTES: usize = 4096; +/// Mirrors the field order `DescribeConfigsRequest::decode` walks. +/// +/// # Errors +/// +/// Returns an error when a declared array or string length cannot fit in the bytes remaining +/// in the frame, or the body is truncated or malformed in a way that cannot be walked. +pub fn validate_describe_configs_shape( Review Comment: this walker and the AlterConfigs one at :1113 have zero tests, and deleting either keeps everything green. They carry real weight: kafka-protocol 0.18 does `Vec::with_capacity(n)` on decode, so an i32::MAX count from a 4-byte body, before auth, aborts the process on allocation. version_firewall_tests.rs:871,896 claim "rejected by the shape walk" but use 100000, which the crate rejects on its own. Add cases with i32::MAX, the max compact varint, nested key/config counts, nullable keys, and v3/flexible layouts. ########## gateways/kafka/docs/CONFIGS.md: ########## @@ -0,0 +1,131 @@ +# Topic configuration + +`DescribeConfigs` (32) and `AlterConfigs` (33) apply to Kafka topic resources only Review Comment: writes only work through legacy API 33, and Kafka 4.0 removed `Admin.alterConfigs`. `kafka-configs.sh --alter` on topics, MirrorMaker 2 and Spring send 44 with no fallback. The later claim (:107-109) that a client sending 44 gets its connection closed is false for Java and librdkafka: both fail client-side, since 44 isn't advertised. Add a client/tool compatibility table and correct that claim. ########## gateways/kafka/src/protocol/handlers/alter_configs.rs: ########## @@ -0,0 +1,325 @@ +// 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. + +//! `AlterConfigs` (API key 33). + +use bytes::Bytes; +use iggy::prelude::IggyError; +use kafka_protocol::messages::AlterConfigsRequest; +use kafka_protocol::messages::alter_configs_request::AlterConfigsResource; +use kafka_protocol::messages::alter_configs_response::{ + AlterConfigsResourceResponse, AlterConfigsResponse, +}; +use kafka_protocol::protocol::StrBytes; +use tokio::time::Instant; + +use crate::bridge::topic_map::validate_kafka_topic_name; +use crate::bridge::{BridgeError, IggyBridge, StreamTopicCache, TopicLoad}; +use crate::protocol::api::{ + API_KEY_ALTER_CONFIGS, ApiVersionRange, ERROR_INVALID_CONFIG, ERROR_INVALID_REQUEST, + ERROR_INVALID_TOPIC_EXCEPTION, ERROR_NONE, ERROR_NOT_CONTROLLER, ERROR_POLICY_VIOLATION, + ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, + REQUEST_DEADLINE, +}; +use crate::protocol::bounds_guard::validate_alter_configs_shape; +use crate::protocol::handlers::topic_config::{ + IggyTopicKey, ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC, bridge_failure, exceeds_topic_cap, + find_duplicate_names, name_reason, plan_retention_update, static_text, topic_cap_message, +}; +use crate::protocol::handlers::{ + decode_guarded, encode_message, is_supported_version, respond_or_close, + unsupported_version_response, +}; + +pub const RANGE: ApiVersionRange = ApiVersionRange { + api_key: API_KEY_ALTER_CONFIGS, + min_version: 0, + max_version: 2, +}; + +pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { + if !is_supported_version(API_KEY_ALTER_CONFIGS, api_version) { + return unsupported_version_response(API_KEY_ALTER_CONFIGS, api_version, |version| { + encode_error_response(version, ERROR_INVALID_REQUEST) + }); + } + + let req = match decode_guarded::<AlterConfigsRequest>(api_version, body, |version, body| { + validate_alter_configs_shape(version, body, state.max_frame_size) + }) { + Ok(req) => req, + Err(error) => { + tracing::debug!(%error, "Failed to decode AlterConfigs request"); + return respond_or_close( + encode_error_response(api_version, ERROR_INVALID_REQUEST), + "AlterConfigs", + ); + } + }; + + let Some(bridge) = &state.bridge else { + let responses = req + .resources + .iter() + .map(|resource| resource_error(resource, ERROR_NOT_CONTROLLER, None)) + .collect(); + return respond_or_close( + encode_message(&response(responses), api_version, 256), + "AlterConfigs", + ); + }; + + let deadline = Instant::now() + REQUEST_DEADLINE; + let responses = alter_all(state, bridge, &req, deadline).await; + respond_or_close( + encode_message(&response(responses), api_version, 256), + "AlterConfigs", + ) +} + +/// # Errors +/// +/// Returns an error when `kafka_protocol` cannot encode the response at `version`. +pub fn encode_error_response(version: i16, error_code: i16) -> crate::error::Result<Bytes> { + let responses = vec![ + AlterConfigsResourceResponse::default() + .with_error_code(error_code) + .with_error_message(None) + .with_resource_type(RESOURCE_TYPE_TOPIC) + .with_resource_name(StrBytes::from_static_str("")), + ]; + encode_message(&response(responses), version, 128) +} + +fn response(responses: Vec<AlterConfigsResourceResponse>) -> AlterConfigsResponse { + AlterConfigsResponse::default() + .with_throttle_time_ms(0) + .with_responses(responses) +} + +async fn alter_all( + state: &GatewayState, + bridge: &IggyBridge, + req: &AlterConfigsRequest, + deadline: Instant, +) -> Vec<AlterConfigsResourceResponse> { + if !req.unknown_tagged_fields.is_empty() { + let message = static_text("unknown tagged fields are not supported"); + return if req.resources.is_empty() { + vec![ + AlterConfigsResourceResponse::default() + .with_error_code(ERROR_INVALID_REQUEST) + .with_error_message(Some(message)) + .with_resource_type(RESOURCE_TYPE_TOPIC) + .with_resource_name(StrBytes::from_static_str("")), + ] + } else { + req.resources + .iter() + .map(|resource| { + resource_error(resource, ERROR_INVALID_REQUEST, Some(message.clone())) + }) + .collect() + }; + } + + let topic_names: Vec<&str> = req + .resources + .iter() + .filter(|resource| resource.resource_type == RESOURCE_TYPE_TOPIC) + .map(|resource| resource.resource_name.as_str()) + .collect(); + if exceeds_topic_cap(topic_names.iter().copied()) { + let message = StrBytes::from(topic_cap_message("AlterConfigs")); + tracing::warn!("AlterConfigs request addresses too many distinct topics; rejecting"); + return req + .resources + .iter() + .map(|resource| resource_error(resource, ERROR_POLICY_VIOLATION, Some(message.clone()))) + .collect(); + } + + // Real Kafka refuses every occurrence of a duplicate resource name outright rather than + // picking a winner - the same choice CreateTopics makes for a repeated topic name. + let duplicate_names = find_duplicate_names(topic_names.iter().copied()); + + let mut cache = StreamTopicCache::default(); + let mut responses = Vec::with_capacity(req.resources.len()); + let mut deadline_exceeded = false; + for resource in &req.resources { + if resource.resource_type == RESOURCE_TYPE_TOPIC + && duplicate_names.contains(resource.resource_name.as_str()) + { + responses.push(resource_error(resource, ERROR_INVALID_REQUEST, None)); + continue; + } + if deadline_exceeded || Instant::now() >= deadline { + if !deadline_exceeded { + deadline_exceeded = true; + tracing::warn!( + "AlterConfigs deadline passed; answering remaining resources retriable \ + instead of starting new Iggy calls" + ); + } + responses.push(resource_error(resource, ERROR_REQUEST_TIMED_OUT, None)); + continue; + } + + let result = alter_one( + state, + bridge, + &mut cache, + resource, + req.validate_only, + deadline, + ) + .await; + if Instant::now() >= deadline { + deadline_exceeded = true; + } + responses.push(result); + } + responses +} + +async fn alter_one( + state: &GatewayState, + bridge: &IggyBridge, + cache: &mut StreamTopicCache, + resource: &AlterConfigsResource, + validate_only: bool, + deadline: Instant, +) -> AlterConfigsResourceResponse { + if !resource.unknown_tagged_fields.is_empty() + || resource + .configs + .iter() + .any(|config| !config.unknown_tagged_fields.is_empty()) + { + return resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(static_text("unknown tagged fields are not supported")), + ); + } + if resource.resource_type != RESOURCE_TYPE_TOPIC { + return resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(static_text(ONLY_TOPIC_RESOURCES)), + ); + } + let name = resource.resource_name.as_str(); + if let Err(error) = validate_kafka_topic_name("kafka_topic", name) { + return resource_error( + resource, + ERROR_INVALID_TOPIC_EXCEPTION, + Some(StrBytes::from(name_reason(&error))), + ); + } + + // Existence is checked before config keys. A missing topic is + // UNKNOWN_TOPIC_OR_PARTITION even when a key would also be invalid. + // A deadline that has already passed must not start the lockstep Iggy call. + match cache.lookup(bridge, name, deadline, "AlterConfigs").await { + TopicLoad::Found(_) => {} + TopicLoad::Missing => { + return resource_error(resource, ERROR_UNKNOWN_TOPIC_OR_PARTITION, None); + } + TopicLoad::Failed(error) => { + let (code, message) = bridge_failure(error, "AlterConfigs", "altering"); + return resource_error(resource, code, message); + } + TopicLoad::NotStarted | TopicLoad::TimedOut => { + return resource_error(resource, ERROR_REQUEST_TIMED_OUT, None); + } + } + + let planned = match plan_retention_update(resource.configs.iter().map(|config| { + ( + config.name.as_str(), + config.value.as_ref().map(StrBytes::as_str), + ) + })) { + Ok(planned) => planned, + Err(fault) => { + return resource_error( + resource, + ERROR_INVALID_CONFIG, + Some(StrBytes::from(fault.message())), + ); + } + }; + // Kafka's AlterConfigs replaces the whole set, so an empty list is a reset. + // This gateway patches named keys. Reporting success would hide that the + // previous expiry is still stored. + let Some(planned) = planned else { + return resource_error( + resource, + ERROR_INVALID_CONFIG, + Some(static_text( + "an empty configs list is rejected because this gateway patches named keys and does not replace the topic configuration", + )), + ); + }; + + // `validate_only` already performed the existence read and the same key checks. + // It does not call `update_topic` and does not change synonym memory. + if validate_only { + return resource_error(resource, ERROR_NONE, None); + } + + // Do not start the write once the deadline has passed, and do not wrap it in Review Comment: this says the write isn't wrapped in a timeout, but `update_kafka_topic_message_expiry` puts `update_topic` inside `with_request_timeout` (bridge/iggy_bridge/topics.rs:312), so the exact case described here still happens. Timeout or TransientNotCommitted answers 7 while a short retention may still commit and delete sealed segments. The handler can also run 20+15 = 35s, past the 25s shutdown drain (api.rs:201) and Java's 30s `request.timeout.ms`. Give the write its own budget inside the drain, fix the comment, and on timeout do one bounded re-read to report the real outcome, or document 7 as outcome unknown in CONFIGS.md. ########## gateways/kafka/src/protocol/handlers/topic_config.rs: ########## @@ -0,0 +1,721 @@ +// 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. + +//! Topic configuration shared by `DescribeConfigs` and `AlterConfigs`. +//! +//! Iggy stores `message_expiry` as [`IggyExpiry`]. A duration is microseconds +//! (`IggyDuration::from(u64)` / `as_micros`). `u64::MAX` is [`IggyExpiry::NeverExpire`] +//! and `0` is [`IggyExpiry::ServerDefault`], so neither is a usable millisecond expiry. +//! Kafka `retention.ms` is a decimal millisecond count, or `-1` when sealed segments +//! never expire. `retention.minutes` and `retention.hours` are the same duration, +//! converted to milliseconds before anything is stored. A positive count is capped +//! at `u32::MAX` seconds, matching [`IggyExpiry::from_str`]. + +use std::collections::{HashMap, HashSet}; +use std::str::FromStr; + +use iggy::prelude::{HeaderKey, IggyDuration, IggyExpiry, ResourceOptions, topic_option_keys}; +use kafka_protocol::protocol::StrBytes; + +use crate::bridge::BridgeError; + +/// Kafka `ConfigResource.Type.TOPIC`. +pub const RESOURCE_TYPE_TOPIC: i8 = 2; + +/// Kafka `ConfigSource.DYNAMIC_TOPIC_CONFIG`: the value was stored on the topic. +pub const CONFIG_SOURCE_TOPIC: i8 = 1; + +/// Kafka `ConfigSource.DEFAULT_CONFIG`: the value is the gateway default. +pub const CONFIG_SOURCE_DEFAULT: i8 = 5; + +/// Kafka `ConfigType.STRING`. +pub const CONFIG_TYPE_STRING: i8 = 2; + +/// Kafka `ConfigType.LONG`. +pub const CONFIG_TYPE_LONG: i8 = 5; + +pub const RETENTION_MS: &str = "retention.ms"; +pub const RETENTION_MINUTES: &str = "retention.minutes"; +pub const RETENTION_HOURS: &str = "retention.hours"; +pub const MILLIS_PER_MINUTE: u64 = 60_000; +pub const MILLIS_PER_HOUR: u64 = 3_600_000; +pub const CLEANUP_POLICY: &str = "cleanup.policy"; +pub const CLEANUP_POLICY_VALUE: &str = "delete"; + +pub const ONLY_TOPIC_RESOURCES: &str = "only topic resources are supported"; + +pub const RETENTION_DOC: &str = "How long a sealed segment is kept, in milliseconds. The active segment does not expire. -1 means sealed segments never expire."; +pub const RETENTION_MINUTES_DOC: &str = "How long a sealed segment is kept, in minutes, derived from retention.ms. The count truncates toward zero. -1 means sealed segments never expire."; +pub const RETENTION_HOURS_DOC: &str = "How long a sealed segment is kept, in hours, derived from retention.ms. The count truncates toward zero. -1 means sealed segments never expire."; +pub const CLEANUP_DOC: &str = "Iggy deletes expired messages and does not compact a topic."; + +/// Distinct topic names one `CreateTopics`, `DescribeConfigs`, or `AlterConfigs` request +/// may address through the bridge. +/// +/// `bounds_guard`'s element ceiling is a pre-decode limit. `CreateTopics` can spend several +/// lockstep Iggy calls per name on the shared client (`ensure_stream`, `create_topic`, and a +/// possible race-retry read). 100 keeps that worst case small and is shared with the config +/// APIs. Duplicate names count once toward the cap and are rejected on their own. A larger +/// batch is `POLICY_VIOLATION` on every resource. +pub const MAX_CONFIG_TOPICS: usize = 100; + +/// One config name a describe request asked for, in request order. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ListedKey { + Retention, + Minutes, + Hours, + Cleanup, + Unknown(String), +} + +/// Why one resource's alter cannot be stored. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ConfigFault { + UnknownKey(String), + Repeated(String), + ReadOnly(&'static str), + InvalidRetention, + ConflictingRetention, +} + +impl ConfigFault { + #[must_use] + pub fn message(&self) -> String { + match self { + Self::UnknownKey(name) => format!("unknown config key '{name}'"), + Self::Repeated(name) => format!("config key '{name}' is repeated"), + Self::ReadOnly(message) => (*message).to_string(), + Self::InvalidRetention => format!( + "retention.ms, retention.minutes, and retention.hours must be -1 or a positive count whose duration is at most {} seconds", + u32::MAX + ), + Self::ConflictingRetention => "retention.ms, retention.minutes, and retention.hours must describe the same duration".to_string(), + } + } +} + +/// `retention.minutes` and `retention.hours` used by the last successful alter. +/// +/// Process-local, keyed by Iggy stream and topic, and not stored in Iggy. A +/// restart clears it. Deleting the topic outside this gateway does not. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct RetentionSynonyms { + pub minutes: bool, + pub hours: bool, +} + +impl RetentionSynonyms { + #[must_use] + pub const fn is_empty(self) -> bool { + !self.minutes && !self.hours + } +} + +/// Iggy stream and topic a Kafka name resolves to. Synonym memory uses this pair, +/// not the Kafka topic name. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct IggyTopicKey { + stream: String, + topic: String, +} + +impl IggyTopicKey { + #[must_use] + pub fn new(stream: impl Into<String>, topic: impl Into<String>) -> Self { + Self { + stream: stream.into(), + topic: topic.into(), + } + } +} + +/// Process memory of [`RetentionSynonyms`] per [`IggyTopicKey`]. +#[derive(Debug, Default)] +pub struct RetentionSynonymMemory { + topics: HashMap<IggyTopicKey, RetentionSynonyms>, +} + +impl RetentionSynonymMemory { + #[must_use] + pub fn remembered(&self, key: &IggyTopicKey) -> RetentionSynonyms { + self.topics.get(key).copied().unwrap_or_default() + } + + /// Remembers `synonyms` after a successful alter that stored an expiry. + /// + /// An empty set clears the key. That is an alter which used only + /// `retention.ms`. A failed alter, and a resource that stored nothing, + /// must not call this. + pub fn record_applied(&mut self, key: IggyTopicKey, synonyms: RetentionSynonyms) { + if synonyms.is_empty() { + self.topics.remove(&key); + } else { + self.topics.insert(key, synonyms); + } + } +} + +/// One duration to store, plus which synonym names the alter used. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PlannedRetention { + pub expiry: IggyExpiry, + pub synonyms: RetentionSynonyms, +} + +/// `retention.ms` as Kafka should report it, plus the config source byte. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RetentionMs { + pub value: String, + pub source: i8, +} + +/// Whether the topic's `message_expiry` option was sent by a client. +/// +/// Admission stores the never-expire default with `explicit == false`. An alter +/// of `retention.ms` stores the key with `explicit == true`, including `-1`. +#[must_use] +pub fn message_expiry_is_explicit(options: &ResourceOptions) -> bool { + let Ok(key) = HeaderKey::from_str(topic_option_keys::MESSAGE_EXPIRY) else { + return false; + }; + options.get(&key).is_some_and(|entry| entry.explicit) +} + +/// Map an Iggy expiry onto Kafka `retention.ms`. +/// +/// A derived never-expire value is [`CONFIG_SOURCE_DEFAULT`]. An explicit one, +/// and any stored duration, is [`CONFIG_SOURCE_TOPIC`]. A duration that is not a +/// positive whole number of milliseconds is refused: reporting a rounded count +/// would describe a different value than Iggy stored. +/// +/// # Errors +/// +/// Returns `Err` when `expiry` is a duration of `0` microseconds or a duration +/// that is not a whole number of milliseconds. +pub fn retention_ms(expiry: IggyExpiry, explicit: bool) -> Result<RetentionMs, ()> { + match expiry { + IggyExpiry::NeverExpire | IggyExpiry::ServerDefault => Ok(RetentionMs { + value: "-1".to_string(), + source: if explicit { + CONFIG_SOURCE_TOPIC + } else { + CONFIG_SOURCE_DEFAULT + }, + }), + IggyExpiry::ExpireDuration(duration) => { + let micros = duration.as_micros(); + if micros == 0 || !micros.is_multiple_of(1_000) { + return Err(()); + } + Ok(RetentionMs { + value: (micros / 1_000).to_string(), + source: CONFIG_SOURCE_TOPIC, + }) + } + } +} + +/// Parse a Kafka `retention.ms` value into the expiry `update_topic` stores. +/// +/// `-1` is [`IggyExpiry::NeverExpire`] (`u64::MAX` on the wire), not +/// [`IggyExpiry::ServerDefault`] (`0`), because the server ignores the `0` +/// sentinel and leaves the previous expiry in place. The text must be the +/// canonical decimal of the parsed integer, so a later describe reports the +/// same characters the alter accepted. +/// +/// # Errors +/// +/// Returns [`ConfigFault::InvalidRetention`] for any value other than `-1` or a +/// positive millisecond count of at most `u32::MAX` seconds. +#[cfg(test)] +pub fn parse_retention_ms(value: &str) -> Result<IggyExpiry, ConfigFault> { + parse_scaled_retention(value, 1) +} + +/// `retention.ms` text scaled into `unit_ms` units. +/// +/// `-1` stays `-1`. Any other count is integer division toward zero, so a stored +/// `90000` reported as minutes is `1` while `retention.ms` stays `90000`. +#[must_use] +pub fn retention_in_unit(millis_text: &str, unit_ms: u64) -> Option<String> { + let Ok(millis) = millis_text.parse::<i64>() else { + return None; + }; + if millis == -1 { + return Some("-1".to_string()); + } + if millis < 0 || unit_ms == 0 { + return None; + } + let Ok(millis) = u64::try_from(millis) else { + return None; + }; + Some((millis / unit_ms).to_string()) +} + +/// `count` is one unit of `unit_ms` milliseconds. `1` is `retention.ms`. +fn parse_scaled_retention(value: &str, unit_ms: u64) -> Result<IggyExpiry, ConfigFault> { + let Ok(count) = value.parse::<i64>() else { + return Err(ConfigFault::InvalidRetention); + }; + if count.to_string() != value { + return Err(ConfigFault::InvalidRetention); + } + if count == -1 { + return Ok(IggyExpiry::NeverExpire); + } + if count <= 0 { + return Err(ConfigFault::InvalidRetention); + } + let count = u64::try_from(count).map_err(|_| ConfigFault::InvalidRetention)?; + // Reject a product that does not fit before multiplying into milliseconds. + if unit_ms == 0 || count > u64::MAX / unit_ms { + return Err(ConfigFault::InvalidRetention); + } + let millis = count * unit_ms; + // `IggyExpiry::from_str` rejects `as_secs() > u32::MAX`. The typed update path + // does not, but a value the server's own parser refuses is not one this gateway + // should store. + if millis / 1_000 > u64::from(u32::MAX) { + return Err(ConfigFault::InvalidRetention); + } + let Some(micros) = millis.checked_mul(1_000) else { + return Err(ConfigFault::InvalidRetention); + }; + Ok(IggyExpiry::ExpireDuration(IggyDuration::from(micros))) +} + +/// Whether `names` contains more than [`MAX_CONFIG_TOPICS`] distinct strings. +#[must_use] +pub fn exceeds_topic_cap<'a>(names: impl IntoIterator<Item = &'a str>) -> bool { + names.into_iter().collect::<HashSet<_>>().len() > MAX_CONFIG_TOPICS +} + +/// Client-facing text for a request over [`MAX_CONFIG_TOPICS`]. +#[must_use] +pub fn topic_cap_message(api_name: &str) -> String { + format!( + "this gateway addresses at most {MAX_CONFIG_TOPICS} distinct topics per {api_name} request" + ) +} + +/// Names in `names` that occur more than once. +/// +/// Real Kafka refuses every occurrence of a duplicate resource name with `INVALID_REQUEST` +/// (42) rather than silently picking a winner - the same choice `CreateTopics`' own +/// `find_duplicate_names` makes for a repeated topic name in one batch. +#[must_use] +pub fn find_duplicate_names<'a>(names: impl IntoIterator<Item = &'a str>) -> HashSet<&'a str> { + let mut seen = HashSet::new(); + let mut duplicates = HashSet::new(); + for name in names { + if !seen.insert(name) { + duplicates.insert(name); + } + } + duplicates +} + +/// The reason a topic name failed Kafka's own naming rules, or a fixed fallback. +#[must_use] +pub fn name_reason(error: &BridgeError) -> String { + match error { + BridgeError::InvalidKafkaTopicName { reason, .. } => reason.clone(), + _ => "invalid topic name".to_string(), + } +} + +#[must_use] +pub const fn static_text(message: &'static str) -> StrBytes { + StrBytes::from_static_str(message) +} + +/// Maps a bridge failure to a Kafka error code and client-facing message, logging the real +/// cause server-side. +/// +/// `handler` names the API in the log line (`"DescribeConfigs"`/`"AlterConfigs"`); `action` is +/// the present participle of what the bridge call was doing (`"reading"`/`"altering"`), reused +/// in both the timeout log and the generic internal-error message. +#[must_use] +pub fn bridge_failure(error: &BridgeError, handler: &str, action: &str) -> (i16, Option<StrBytes>) { + let code = error.to_kafka_error_code(); + match error { + BridgeError::InvalidKafkaTopicName { reason, .. } => { + tracing::debug!(reason, "{handler} rejected an invalid topic name"); + (code, Some(StrBytes::from(reason.clone()))) + } + BridgeError::Timeout | BridgeError::SendLost(_) => { + tracing::warn!(%error, "{handler} {action} exceeded the bridge deadline"); + (code, None) + } + other => { + tracing::error!(%other, "{handler} failed while {action} topic configuration"); + ( + code, + Some(StrBytes::from(format!( + "internal error {action} topic configuration" + ))), + ) + } + } +} + +/// Keys to return for one describe resource. +/// +/// `None` and an empty list both mean `retention.ms` and `cleanup.policy`. +/// `retention.minutes` and `retention.hours` are listed only when the client +/// names them. Any other name is [`ListedKey::Unknown`]. `DescribeConfigs` omits +/// those names from the response and does not fail the resource. +#[must_use] +pub fn listed_keys(configuration_keys: Option<&[StrBytes]>) -> Vec<ListedKey> { + let Some(keys) = configuration_keys.filter(|keys| !keys.is_empty()) else { + return vec![ListedKey::Retention, ListedKey::Cleanup]; + }; + keys.iter() + .map(|key| match key.as_str() { + RETENTION_MS => ListedKey::Retention, + RETENTION_MINUTES => ListedKey::Minutes, + RETENTION_HOURS => ListedKey::Hours, + CLEANUP_POLICY => ListedKey::Cleanup, + other => ListedKey::Unknown(other.to_string()), + }) + .collect() +} + +/// The expiry to store, or `None` when the resource names no retention key. +/// +/// `retention.ms`, `retention.minutes`, and `retention.hours` are one duration. +/// Minutes and hours convert to milliseconds. Two names that convert to +/// different millisecond values fail the resource and select nothing. One +/// expiry is returned when they agree. [`PlannedRetention::synonyms`] is empty +/// when the resource used only `retention.ms`. +/// +/// One invalid or repeated key fails the whole resource. Nothing is selected +/// for storage in that case. `update_topic` stores the millisecond expiry, never +/// the minute or hour count. +/// +/// # Errors +/// +/// Returns the first [`ConfigFault`] in request order. +pub fn plan_retention_update<'a>( + configs: impl IntoIterator<Item = (&'a str, Option<&'a str>)>, +) -> Result<Option<PlannedRetention>, ConfigFault> { + let mut seen = HashSet::new(); + let mut expiry = None; + let mut synonyms = RetentionSynonyms::default(); + for (name, value) in configs { + if !seen.insert(name) { + return Err(ConfigFault::Repeated(name.to_string())); + } + let unit_ms = retention_scale(name, &mut synonyms)?; + let Some(value) = value else { + return Err(ConfigFault::InvalidRetention); + }; + let parsed = parse_scaled_retention(value, unit_ms)?; + if let Some(existing) = expiry + && existing != parsed + { + return Err(ConfigFault::ConflictingRetention); + } + expiry = Some(parsed); + } + Ok(expiry.map(|expiry| PlannedRetention { expiry, synonyms })) +} + +fn retention_scale(name: &str, synonyms: &mut RetentionSynonyms) -> Result<u64, ConfigFault> { Review Comment: `retention.minutes` / `retention.hours` aren't Kafka topic configs. Only broker `log.retention.*` exists, and a broker answers 40 "Unknown topic config name" for them. Describe then reports them as DYNAMIC_TOPIC_CONFIG synonyms from `RetentionSynonymMemory` (:112), which is per-instance process memory that is lost on restart and stale after any external change. Reject both with 40 and drop the memory. If synonyms are wanted, report `retention.ms` itself, Kafka style. ########## gateways/kafka/src/protocol/handlers/topic_config.rs: ########## @@ -0,0 +1,721 @@ +// 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. + +//! Topic configuration shared by `DescribeConfigs` and `AlterConfigs`. +//! +//! Iggy stores `message_expiry` as [`IggyExpiry`]. A duration is microseconds +//! (`IggyDuration::from(u64)` / `as_micros`). `u64::MAX` is [`IggyExpiry::NeverExpire`] +//! and `0` is [`IggyExpiry::ServerDefault`], so neither is a usable millisecond expiry. +//! Kafka `retention.ms` is a decimal millisecond count, or `-1` when sealed segments +//! never expire. `retention.minutes` and `retention.hours` are the same duration, +//! converted to milliseconds before anything is stored. A positive count is capped +//! at `u32::MAX` seconds, matching [`IggyExpiry::from_str`]. + +use std::collections::{HashMap, HashSet}; +use std::str::FromStr; + +use iggy::prelude::{HeaderKey, IggyDuration, IggyExpiry, ResourceOptions, topic_option_keys}; +use kafka_protocol::protocol::StrBytes; + +use crate::bridge::BridgeError; + +/// Kafka `ConfigResource.Type.TOPIC`. +pub const RESOURCE_TYPE_TOPIC: i8 = 2; + +/// Kafka `ConfigSource.DYNAMIC_TOPIC_CONFIG`: the value was stored on the topic. +pub const CONFIG_SOURCE_TOPIC: i8 = 1; + +/// Kafka `ConfigSource.DEFAULT_CONFIG`: the value is the gateway default. +pub const CONFIG_SOURCE_DEFAULT: i8 = 5; + +/// Kafka `ConfigType.STRING`. +pub const CONFIG_TYPE_STRING: i8 = 2; + +/// Kafka `ConfigType.LONG`. +pub const CONFIG_TYPE_LONG: i8 = 5; + +pub const RETENTION_MS: &str = "retention.ms"; +pub const RETENTION_MINUTES: &str = "retention.minutes"; +pub const RETENTION_HOURS: &str = "retention.hours"; +pub const MILLIS_PER_MINUTE: u64 = 60_000; +pub const MILLIS_PER_HOUR: u64 = 3_600_000; +pub const CLEANUP_POLICY: &str = "cleanup.policy"; +pub const CLEANUP_POLICY_VALUE: &str = "delete"; + +pub const ONLY_TOPIC_RESOURCES: &str = "only topic resources are supported"; + +pub const RETENTION_DOC: &str = "How long a sealed segment is kept, in milliseconds. The active segment does not expire. -1 means sealed segments never expire."; +pub const RETENTION_MINUTES_DOC: &str = "How long a sealed segment is kept, in minutes, derived from retention.ms. The count truncates toward zero. -1 means sealed segments never expire."; +pub const RETENTION_HOURS_DOC: &str = "How long a sealed segment is kept, in hours, derived from retention.ms. The count truncates toward zero. -1 means sealed segments never expire."; +pub const CLEANUP_DOC: &str = "Iggy deletes expired messages and does not compact a topic."; + +/// Distinct topic names one `CreateTopics`, `DescribeConfigs`, or `AlterConfigs` request +/// may address through the bridge. +/// +/// `bounds_guard`'s element ceiling is a pre-decode limit. `CreateTopics` can spend several +/// lockstep Iggy calls per name on the shared client (`ensure_stream`, `create_topic`, and a +/// possible race-retry read). 100 keeps that worst case small and is shared with the config +/// APIs. Duplicate names count once toward the cap and are rejected on their own. A larger +/// batch is `POLICY_VIOLATION` on every resource. +pub const MAX_CONFIG_TOPICS: usize = 100; + +/// One config name a describe request asked for, in request order. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ListedKey { + Retention, + Minutes, + Hours, + Cleanup, + Unknown(String), +} + +/// Why one resource's alter cannot be stored. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ConfigFault { + UnknownKey(String), + Repeated(String), + ReadOnly(&'static str), + InvalidRetention, + ConflictingRetention, +} + +impl ConfigFault { + #[must_use] + pub fn message(&self) -> String { + match self { + Self::UnknownKey(name) => format!("unknown config key '{name}'"), + Self::Repeated(name) => format!("config key '{name}' is repeated"), + Self::ReadOnly(message) => (*message).to_string(), + Self::InvalidRetention => format!( + "retention.ms, retention.minutes, and retention.hours must be -1 or a positive count whose duration is at most {} seconds", + u32::MAX + ), + Self::ConflictingRetention => "retention.ms, retention.minutes, and retention.hours must describe the same duration".to_string(), + } + } +} + +/// `retention.minutes` and `retention.hours` used by the last successful alter. +/// +/// Process-local, keyed by Iggy stream and topic, and not stored in Iggy. A +/// restart clears it. Deleting the topic outside this gateway does not. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct RetentionSynonyms { + pub minutes: bool, + pub hours: bool, +} + +impl RetentionSynonyms { + #[must_use] + pub const fn is_empty(self) -> bool { + !self.minutes && !self.hours + } +} + +/// Iggy stream and topic a Kafka name resolves to. Synonym memory uses this pair, +/// not the Kafka topic name. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct IggyTopicKey { + stream: String, + topic: String, +} + +impl IggyTopicKey { + #[must_use] + pub fn new(stream: impl Into<String>, topic: impl Into<String>) -> Self { + Self { + stream: stream.into(), + topic: topic.into(), + } + } +} + +/// Process memory of [`RetentionSynonyms`] per [`IggyTopicKey`]. +#[derive(Debug, Default)] +pub struct RetentionSynonymMemory { + topics: HashMap<IggyTopicKey, RetentionSynonyms>, +} + +impl RetentionSynonymMemory { + #[must_use] + pub fn remembered(&self, key: &IggyTopicKey) -> RetentionSynonyms { + self.topics.get(key).copied().unwrap_or_default() + } + + /// Remembers `synonyms` after a successful alter that stored an expiry. + /// + /// An empty set clears the key. That is an alter which used only + /// `retention.ms`. A failed alter, and a resource that stored nothing, + /// must not call this. + pub fn record_applied(&mut self, key: IggyTopicKey, synonyms: RetentionSynonyms) { + if synonyms.is_empty() { + self.topics.remove(&key); + } else { + self.topics.insert(key, synonyms); + } + } +} + +/// One duration to store, plus which synonym names the alter used. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PlannedRetention { + pub expiry: IggyExpiry, + pub synonyms: RetentionSynonyms, +} + +/// `retention.ms` as Kafka should report it, plus the config source byte. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RetentionMs { + pub value: String, + pub source: i8, +} + +/// Whether the topic's `message_expiry` option was sent by a client. +/// +/// Admission stores the never-expire default with `explicit == false`. An alter +/// of `retention.ms` stores the key with `explicit == true`, including `-1`. +#[must_use] +pub fn message_expiry_is_explicit(options: &ResourceOptions) -> bool { + let Ok(key) = HeaderKey::from_str(topic_option_keys::MESSAGE_EXPIRY) else { + return false; + }; + options.get(&key).is_some_and(|entry| entry.explicit) +} + +/// Map an Iggy expiry onto Kafka `retention.ms`. +/// +/// A derived never-expire value is [`CONFIG_SOURCE_DEFAULT`]. An explicit one, +/// and any stored duration, is [`CONFIG_SOURCE_TOPIC`]. A duration that is not a +/// positive whole number of milliseconds is refused: reporting a rounded count +/// would describe a different value than Iggy stored. +/// +/// # Errors +/// +/// Returns `Err` when `expiry` is a duration of `0` microseconds or a duration +/// that is not a whole number of milliseconds. +pub fn retention_ms(expiry: IggyExpiry, explicit: bool) -> Result<RetentionMs, ()> { + match expiry { + IggyExpiry::NeverExpire | IggyExpiry::ServerDefault => Ok(RetentionMs { + value: "-1".to_string(), + source: if explicit { + CONFIG_SOURCE_TOPIC + } else { + CONFIG_SOURCE_DEFAULT + }, + }), + IggyExpiry::ExpireDuration(duration) => { + let micros = duration.as_micros(); + if micros == 0 || !micros.is_multiple_of(1_000) { + return Err(()); + } + Ok(RetentionMs { + value: (micros / 1_000).to_string(), + source: CONFIG_SOURCE_TOPIC, + }) + } + } +} + +/// Parse a Kafka `retention.ms` value into the expiry `update_topic` stores. +/// +/// `-1` is [`IggyExpiry::NeverExpire`] (`u64::MAX` on the wire), not +/// [`IggyExpiry::ServerDefault`] (`0`), because the server ignores the `0` +/// sentinel and leaves the previous expiry in place. The text must be the +/// canonical decimal of the parsed integer, so a later describe reports the +/// same characters the alter accepted. +/// +/// # Errors +/// +/// Returns [`ConfigFault::InvalidRetention`] for any value other than `-1` or a +/// positive millisecond count of at most `u32::MAX` seconds. +#[cfg(test)] +pub fn parse_retention_ms(value: &str) -> Result<IggyExpiry, ConfigFault> { + parse_scaled_retention(value, 1) +} + +/// `retention.ms` text scaled into `unit_ms` units. +/// +/// `-1` stays `-1`. Any other count is integer division toward zero, so a stored +/// `90000` reported as minutes is `1` while `retention.ms` stays `90000`. +#[must_use] +pub fn retention_in_unit(millis_text: &str, unit_ms: u64) -> Option<String> { + let Ok(millis) = millis_text.parse::<i64>() else { + return None; + }; + if millis == -1 { + return Some("-1".to_string()); + } + if millis < 0 || unit_ms == 0 { + return None; + } + let Ok(millis) = u64::try_from(millis) else { + return None; + }; + Some((millis / unit_ms).to_string()) +} + +/// `count` is one unit of `unit_ms` milliseconds. `1` is `retention.ms`. +fn parse_scaled_retention(value: &str, unit_ms: u64) -> Result<IggyExpiry, ConfigFault> { + let Ok(count) = value.parse::<i64>() else { + return Err(ConfigFault::InvalidRetention); + }; + if count.to_string() != value { + return Err(ConfigFault::InvalidRetention); + } + if count == -1 { + return Ok(IggyExpiry::NeverExpire); + } + if count <= 0 { + return Err(ConfigFault::InvalidRetention); + } + let count = u64::try_from(count).map_err(|_| ConfigFault::InvalidRetention)?; + // Reject a product that does not fit before multiplying into milliseconds. + if unit_ms == 0 || count > u64::MAX / unit_ms { + return Err(ConfigFault::InvalidRetention); + } + let millis = count * unit_ms; + // `IggyExpiry::from_str` rejects `as_secs() > u32::MAX`. The typed update path + // does not, but a value the server's own parser refuses is not one this gateway + // should store. + if millis / 1_000 > u64::from(u32::MAX) { + return Err(ConfigFault::InvalidRetention); + } + let Some(micros) = millis.checked_mul(1_000) else { + return Err(ConfigFault::InvalidRetention); + }; + Ok(IggyExpiry::ExpireDuration(IggyDuration::from(micros))) +} + +/// Whether `names` contains more than [`MAX_CONFIG_TOPICS`] distinct strings. +#[must_use] +pub fn exceeds_topic_cap<'a>(names: impl IntoIterator<Item = &'a str>) -> bool { Review Comment: none of the guards are pinned. These mutations stay green: dropping this 100-topic cap, the deadline-before-write check (alter :290), the tagged-field reject and record-synonyms. The cap also sits after the bridge-off return (describe_configs.rs:80, alter_configs.rs:74), so no serverless test can reach it. Move the cap check before the bridge check and add 100/101 boundary tests. -- 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]
