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