ryerraguntla commented on code in PR #4398: URL: https://github.com/apache/iggy/pull/4398#discussion_r4205917605
########## 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 { Review Comment: fixed ########## 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: fixed ########## 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: fixed ########## 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> { Review Comment: fixed ########## 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 { Review Comment: fixed ########## 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> { 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]
