ryerraguntla commented on code in PR #4398: URL: https://github.com/apache/iggy/pull/4398#discussion_r4179488236
########## gateways/kafka/tests/configs_java_client_tests.rs: ########## @@ -0,0 +1,290 @@ +// 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. + +//! Kafka 3.9 `AdminClient` describe/alter against the gateway, on the host JDK. +//! +//! The docker `--network host` suite cannot reach a loopback gateway from macOS. +//! This test publishes nothing into a container: it binds `127.0.0.1` and runs +//! `java` on the host. Jars come from Maven Central when they are not cached. + +use std::io::Read; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output, Stdio}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use tokio::sync::broadcast; + +use iggy_gateway_kafka::bridge::IggyBridge; +use iggy_gateway_kafka::server::bind_listener; +use iggy_gateway_kafka::{GatewayConfig, KafkaGateway}; + +#[path = "common/iggy_server.rs"] +mod iggy_server; + +use iggy_server::TestServer; + +const TOPIC: &str = "orders"; + +/// Bounds a child the way `kafka_client_e2e_tests.rs` bounds a client run, without GNU `timeout` +/// (this suite runs on the host JDK, including macOS). A hung `java` or `curl` used to block the +/// process until the runner itself was killed. +const CHILD_LIMIT: Duration = Duration::from_secs(120); + +/// Reports why the suite cannot run, and whether that is fatal. +/// +/// Returns `true` when the caller should skip. `JAVA_E2E_REQUIRED=1` makes it panic instead, so a +/// CI job that means to run this fails loudly rather than reporting a pass over zero assertions - +/// same convention as `kafka_client_e2e_tests.rs`'s own `skip`/`KAFKA_E2E_REQUIRED`. +fn skip(reason: &str) -> bool { + assert!( + std::env::var("JAVA_E2E_REQUIRED").as_deref() != Ok("1"), + "JAVA_E2E_REQUIRED=1 but the suite cannot run: {reason}" + ); + eprintln!("skipping Java admin-client end-to-end test: {reason}"); + true +} + +fn java_missing() -> bool { + let available = Command::new("java") + .arg("-version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|status| status.success()); + if available { + false + } else { + skip("java is unavailable") + } +} + +fn output_bounded(command: &mut Command, what: &str) -> Output { + let mut child = command + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap_or_else(|error| panic!("spawn {what}: {error}")); + let mut stdout = child.stdout.take().expect("stdout"); + let mut stderr = child.stderr.take().expect("stderr"); + let stdout_thread = std::thread::spawn(move || { + let mut buf = Vec::new(); + stdout.read_to_end(&mut buf).map(|_read| buf) + }); + let stderr_thread = std::thread::spawn(move || { + let mut buf = Vec::new(); + stderr.read_to_end(&mut buf).map(|_read| buf) + }); + let deadline = Instant::now() + CHILD_LIMIT; + let status = loop { + match child.try_wait() { + Ok(Some(status)) => break status, + Ok(None) if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + panic!("{what} exceeded {CHILD_LIMIT:?}"); + } + Ok(None) => std::thread::sleep(Duration::from_millis(50)), + Err(error) => panic!("poll {what}: {error}"), + } + }; + let stdout = stdout_thread + .join() + .unwrap_or_else(|_| panic!("{what} stdout thread")) + .unwrap_or_else(|error| panic!("read {what} stdout: {error}")); + let stderr = stderr_thread + .join() + .unwrap_or_else(|_| panic!("{what} stderr thread")) + .unwrap_or_else(|error| panic!("read {what} stderr: {error}")); + Output { + status, + stdout, + stderr, + } +} + +const JARS: &[(&str, &str)] = &[ + ( + "kafka-clients-3.9.0.jar", + "https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.9.0/kafka-clients-3.9.0.jar";, + ), + ( + "slf4j-api-1.7.36.jar", + "https://repo1.maven.org/maven2/org/slf4j/slf4j-api/1.7.36/slf4j-api-1.7.36.jar";, + ), + ( + "slf4j-simple-1.7.36.jar", + "https://repo1.maven.org/maven2/org/slf4j/slf4j-simple/1.7.36/slf4j-simple-1.7.36.jar";, + ), + ( + "lz4-java-1.8.0.jar", + "https://repo1.maven.org/maven2/org/lz4/lz4-java/1.8.0/lz4-java-1.8.0.jar";, + ), + ( + "snappy-java-1.1.10.5.jar", + "https://repo1.maven.org/maven2/org/xerial/snappy/snappy-java/1.1.10.5/snappy-java-1.1.10.5.jar";, + ), + ( + "zstd-jni-1.5.6-4.jar", + "https://repo1.maven.org/maven2/com/github/luben/zstd-jni/1.5.6-4/zstd-jni-1.5.6-4.jar";, + ), +]; + +fn jar_dir() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../target/kafka-3.9-admin-jars") +} + +fn ensure_jars(dir: &Path) { + std::fs::create_dir_all(dir).expect("create jar cache"); + for (name, url) in JARS { + let path = dir.join(name); + if path.exists() { + continue; + } + let output = output_bounded( + Command::new("curl") + .args([ + "--fail", + "--silent", + "--show-error", + "--location", + "--max-time", + "60", + "-o", + ]) + .arg(&path) + .arg(url), + name, + ); + if !output.status.success() { + let _ = std::fs::remove_file(&path); + } + assert!( Review Comment: fixed ########## gateways/kafka/src/protocol/handlers/topic_config.rs: ########## @@ -0,0 +1,725 @@ +// 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; + +/// Kafka `ConfigType.UNKNOWN`. +pub const CONFIG_TYPE_UNKNOWN: i8 = 0; + +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 Kafka topic names one `DescribeConfigs` or `AlterConfigs` request may send to Iggy. +/// +/// Same ceiling as `CreateTopics`: each distinct name is a lockstep call on the shared client. +/// Duplicate names in one request count once for this cap, but a repeated name is itself +/// rejected outright - see [`find_duplicate_names`]. A larger batch is `POLICY_VIOLATION` on +/// every resource. The decode-time element ceiling is a separate, larger limit and is not this +/// cap. +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. +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`] and stays in the list. +#[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 = if unit_ms == 1 { 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]
