github-actions[bot] commented on code in PR #4398: URL: https://github.com/apache/iggy/pull/4398#discussion_r4179272512
########## gateways/kafka/src/protocol/handlers/alter_configs.rs: ########## @@ -0,0 +1,320 @@ +// 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}; +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 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, 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, + 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. + if Instant::now() >= deadline { + tracing::warn!( + resource = name, + "AlterConfigs deadline passed before the topic lookup" + ); + return resource_error(resource, ERROR_REQUEST_TIMED_OUT, None); + } + match tokio::time::timeout_at(deadline, bridge.get_kafka_topic(name)).await { + Ok(Ok(Some(_topic))) => {} + Ok(Ok(None)) => { + return resource_error(resource, ERROR_UNKNOWN_TOPIC_OR_PARTITION, None); + } + Ok(Err(error)) => { + let (code, message) = bridge_failure(&error, "AlterConfigs", "altering"); + return resource_error(resource, code, message); + } + Err(_elapsed) => { + tracing::warn!( + resource = name, + "AlterConfigs: this resource's bridge work exceeded the request deadline; \ + answering retriable instead of blocking further" + ); + 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())), + ); + } + }; + // A resource that names no retention key has nothing to store. Omitting the key + // does not clear a previously set expiry or the remembered synonym names. + let Some(planned) = planned else { Review Comment: warning: An empty `configs` list makes `AlterConfigs` answer `ERROR_NONE` and write nothing, but Kafka's legacy `AlterConfigs` replaces the full config set, so a client that asks for a reset sees success. Reject the empty list with `INVALID_CONFIG` (40), and state the replace-to-patch deviation in `docs/CONFIGS.md`. ########## 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: nit: `ensure_jars()` panics when `curl` fails, even with `JAVA_E2E_REQUIRED` unset, so an offline machine with a JDK fails the suite instead of skipping. Make it report the failure so the caller can call `skip()`. ########## gateways/kafka/src/protocol/handlers/describe_configs.rs: ########## @@ -0,0 +1,681 @@ +// 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. + +//! `DescribeConfigs` (API key 32). + +use bytes::Bytes; +use iggy::prelude::TopicDetails; +use kafka_protocol::messages::DescribeConfigsRequest; +use kafka_protocol::messages::describe_configs_request::DescribeConfigsResource; +use kafka_protocol::messages::describe_configs_response::{ + DescribeConfigsResourceResult, DescribeConfigsResponse, DescribeConfigsResult, + DescribeConfigsSynonym, +}; +use kafka_protocol::protocol::StrBytes; +use tokio::time::Instant; + +use crate::bridge::IggyBridge; +use crate::bridge::topic_map::validate_kafka_topic_name; +use crate::protocol::api::{ + API_KEY_DESCRIBE_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_describe_configs_shape; +use crate::protocol::handlers::topic_config::{ + CLEANUP_DOC, CLEANUP_POLICY, CLEANUP_POLICY_VALUE, CONFIG_SOURCE_DEFAULT, CONFIG_SOURCE_TOPIC, + CONFIG_TYPE_LONG, CONFIG_TYPE_STRING, CONFIG_TYPE_UNKNOWN, IggyTopicKey, ListedKey, + MILLIS_PER_HOUR, MILLIS_PER_MINUTE, ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC, RETENTION_DOC, + RETENTION_HOURS, RETENTION_HOURS_DOC, RETENTION_MINUTES, RETENTION_MINUTES_DOC, RETENTION_MS, + RetentionMs, RetentionSynonyms, bridge_failure, exceeds_topic_cap, find_duplicate_names, + listed_keys, message_expiry_is_explicit, name_reason, retention_in_unit, retention_ms, + 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_DESCRIBE_CONFIGS, + min_version: 1, + max_version: 4, +}; + +pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { + if !is_supported_version(API_KEY_DESCRIBE_CONFIGS, api_version) { + return unsupported_version_response(API_KEY_DESCRIBE_CONFIGS, api_version, |version| { + encode_error_response(version, ERROR_INVALID_REQUEST) + }); + } + + let req = match decode_guarded::<DescribeConfigsRequest>(api_version, body, |version, body| { + validate_describe_configs_shape(version, body, state.max_frame_size) + }) { + Ok(req) => req, + Err(error) => { + tracing::debug!(%error, "Failed to decode DescribeConfigs request"); + return respond_or_close( + encode_error_response(api_version, ERROR_INVALID_REQUEST), + "DescribeConfigs", + ); + } + }; + + let Some(bridge) = &state.bridge else { + let results = req + .resources + .iter() + .map(|resource| resource_error(resource, ERROR_NOT_CONTROLLER, None, Vec::new())) + .collect(); + return respond_or_close( + encode_message(&response(results), api_version, 256), + "DescribeConfigs", + ); + }; + + let deadline = Instant::now() + REQUEST_DEADLINE; + let results = describe_all(state, bridge, &req, deadline).await; + respond_or_close( + encode_message(&response(results), api_version, 256), + "DescribeConfigs", + ) +} + +/// # 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 results = vec![ + DescribeConfigsResult::default() + .with_error_code(error_code) + .with_error_message(None) + .with_resource_type(RESOURCE_TYPE_TOPIC) + .with_resource_name(StrBytes::from_static_str("")) + .with_configs(Vec::new()), + ]; + encode_message(&response(results), version, 128) +} + +fn response(results: Vec<DescribeConfigsResult>) -> DescribeConfigsResponse { + DescribeConfigsResponse::default() + .with_throttle_time_ms(0) + .with_results(results) +} + +async fn describe_all( + state: &GatewayState, + bridge: &IggyBridge, + req: &DescribeConfigsRequest, + deadline: Instant, +) -> Vec<DescribeConfigsResult> { + if !req.unknown_tagged_fields.is_empty() { + let message = static_text("unknown tagged fields are not supported"); + return if req.resources.is_empty() { + vec![ + DescribeConfigsResult::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("")) + .with_configs(Vec::new()), + ] + } else { + req.resources + .iter() + .map(|resource| { + resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(message.clone()), + Vec::new(), + ) + }) + .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("DescribeConfigs")); + tracing::warn!("DescribeConfigs request addresses too many distinct topics; rejecting"); + return req + .resources + .iter() + .map(|resource| { + resource_error( + resource, + ERROR_POLICY_VIOLATION, + Some(message.clone()), + Vec::new(), + ) + }) + .collect(); + } + + // Real Kafka refuses every occurrence of a duplicate resource name outright rather than + // answering it from a cached result that may have been filtered by a different + // `configuration_keys` list - the same choice CreateTopics makes for a repeated topic name. + let duplicate_names = find_duplicate_names(topic_names.iter().copied()); + + let mut results = 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()) + { + results.push(resource_error( + resource, + ERROR_INVALID_REQUEST, + None, + Vec::new(), + )); + continue; + } + if deadline_exceeded || Instant::now() >= deadline { + if !deadline_exceeded { + deadline_exceeded = true; + tracing::warn!( + "DescribeConfigs deadline passed; answering remaining resources retriable \ + instead of starting new Iggy calls" + ); + } + results.push(resource_error( + resource, + ERROR_REQUEST_TIMED_OUT, + None, + Vec::new(), + )); + continue; + } + + let result = describe_one( + state, + bridge, + resource, + req.include_synonyms, + req.include_documentation, + deadline, + ) + .await; + if Instant::now() >= deadline { + deadline_exceeded = true; + } + results.push(result); + } + results +} + +async fn describe_one( + state: &GatewayState, + bridge: &IggyBridge, + resource: &DescribeConfigsResource, + include_synonyms: bool, + include_documentation: bool, + deadline: Instant, +) -> DescribeConfigsResult { + if !resource.unknown_tagged_fields.is_empty() { + return resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(static_text("unknown tagged fields are not supported")), + Vec::new(), + ); + } + if resource.resource_type != RESOURCE_TYPE_TOPIC { + return resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(static_text(ONLY_TOPIC_RESOURCES)), + Vec::new(), + ); + } + 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))), + Vec::new(), + ); + } + + if Instant::now() >= deadline { + tracing::warn!( + resource = name, + "DescribeConfigs deadline passed before the topic lookup" + ); + return resource_error(resource, ERROR_REQUEST_TIMED_OUT, None, Vec::new()); + } + let topic = match tokio::time::timeout_at(deadline, bridge.get_kafka_topic(name)).await { Review Comment: warning: The resource loop pays one `get_kafka_topic()` round trip per resource, so a request naming 100 topics makes 100 sequential calls on the shared lockstep client, and each one blocks Produce and Metadata. Group resources by resolved stream and call `get_topics` once per stream, as `get_kafka_topics()` does. ########## 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; Review Comment: simplification: `MAX_CONFIG_TOPICS` repeats the 100-topic cap that `create_topics.rs` defines, and `topic_cap_message()` repeats its message. Declare the cap and message once and use them in both modules. ########## gateways/kafka/src/protocol/handlers/describe_configs.rs: ########## @@ -0,0 +1,681 @@ +// 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. + +//! `DescribeConfigs` (API key 32). + +use bytes::Bytes; +use iggy::prelude::TopicDetails; +use kafka_protocol::messages::DescribeConfigsRequest; +use kafka_protocol::messages::describe_configs_request::DescribeConfigsResource; +use kafka_protocol::messages::describe_configs_response::{ + DescribeConfigsResourceResult, DescribeConfigsResponse, DescribeConfigsResult, + DescribeConfigsSynonym, +}; +use kafka_protocol::protocol::StrBytes; +use tokio::time::Instant; + +use crate::bridge::IggyBridge; +use crate::bridge::topic_map::validate_kafka_topic_name; +use crate::protocol::api::{ + API_KEY_DESCRIBE_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_describe_configs_shape; +use crate::protocol::handlers::topic_config::{ + CLEANUP_DOC, CLEANUP_POLICY, CLEANUP_POLICY_VALUE, CONFIG_SOURCE_DEFAULT, CONFIG_SOURCE_TOPIC, + CONFIG_TYPE_LONG, CONFIG_TYPE_STRING, CONFIG_TYPE_UNKNOWN, IggyTopicKey, ListedKey, + MILLIS_PER_HOUR, MILLIS_PER_MINUTE, ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC, RETENTION_DOC, + RETENTION_HOURS, RETENTION_HOURS_DOC, RETENTION_MINUTES, RETENTION_MINUTES_DOC, RETENTION_MS, + RetentionMs, RetentionSynonyms, bridge_failure, exceeds_topic_cap, find_duplicate_names, + listed_keys, message_expiry_is_explicit, name_reason, retention_in_unit, retention_ms, + 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_DESCRIBE_CONFIGS, + min_version: 1, + max_version: 4, +}; + +pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { + if !is_supported_version(API_KEY_DESCRIBE_CONFIGS, api_version) { + return unsupported_version_response(API_KEY_DESCRIBE_CONFIGS, api_version, |version| { + encode_error_response(version, ERROR_INVALID_REQUEST) + }); + } + + let req = match decode_guarded::<DescribeConfigsRequest>(api_version, body, |version, body| { + validate_describe_configs_shape(version, body, state.max_frame_size) + }) { + Ok(req) => req, + Err(error) => { + tracing::debug!(%error, "Failed to decode DescribeConfigs request"); + return respond_or_close( + encode_error_response(api_version, ERROR_INVALID_REQUEST), + "DescribeConfigs", + ); + } + }; + + let Some(bridge) = &state.bridge else { + let results = req + .resources + .iter() + .map(|resource| resource_error(resource, ERROR_NOT_CONTROLLER, None, Vec::new())) + .collect(); + return respond_or_close( + encode_message(&response(results), api_version, 256), + "DescribeConfigs", + ); + }; + + let deadline = Instant::now() + REQUEST_DEADLINE; + let results = describe_all(state, bridge, &req, deadline).await; + respond_or_close( + encode_message(&response(results), api_version, 256), + "DescribeConfigs", + ) +} + +/// # 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 results = vec![ + DescribeConfigsResult::default() + .with_error_code(error_code) + .with_error_message(None) + .with_resource_type(RESOURCE_TYPE_TOPIC) + .with_resource_name(StrBytes::from_static_str("")) + .with_configs(Vec::new()), + ]; + encode_message(&response(results), version, 128) +} + +fn response(results: Vec<DescribeConfigsResult>) -> DescribeConfigsResponse { + DescribeConfigsResponse::default() + .with_throttle_time_ms(0) + .with_results(results) +} + +async fn describe_all( + state: &GatewayState, + bridge: &IggyBridge, + req: &DescribeConfigsRequest, + deadline: Instant, +) -> Vec<DescribeConfigsResult> { + if !req.unknown_tagged_fields.is_empty() { + let message = static_text("unknown tagged fields are not supported"); + return if req.resources.is_empty() { + vec![ + DescribeConfigsResult::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("")) + .with_configs(Vec::new()), + ] + } else { + req.resources + .iter() + .map(|resource| { + resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(message.clone()), + Vec::new(), + ) + }) + .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("DescribeConfigs")); + tracing::warn!("DescribeConfigs request addresses too many distinct topics; rejecting"); + return req + .resources + .iter() + .map(|resource| { + resource_error( + resource, + ERROR_POLICY_VIOLATION, + Some(message.clone()), + Vec::new(), + ) + }) + .collect(); + } + + // Real Kafka refuses every occurrence of a duplicate resource name outright rather than + // answering it from a cached result that may have been filtered by a different + // `configuration_keys` list - the same choice CreateTopics makes for a repeated topic name. + let duplicate_names = find_duplicate_names(topic_names.iter().copied()); + + let mut results = 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()) + { + results.push(resource_error( + resource, + ERROR_INVALID_REQUEST, + None, + Vec::new(), + )); + continue; + } + if deadline_exceeded || Instant::now() >= deadline { + if !deadline_exceeded { + deadline_exceeded = true; + tracing::warn!( + "DescribeConfigs deadline passed; answering remaining resources retriable \ + instead of starting new Iggy calls" + ); + } + results.push(resource_error( + resource, + ERROR_REQUEST_TIMED_OUT, + None, + Vec::new(), + )); + continue; + } + + let result = describe_one( + state, + bridge, + resource, + req.include_synonyms, + req.include_documentation, + deadline, + ) + .await; + if Instant::now() >= deadline { + deadline_exceeded = true; + } + results.push(result); + } + results +} + +async fn describe_one( + state: &GatewayState, + bridge: &IggyBridge, + resource: &DescribeConfigsResource, + include_synonyms: bool, + include_documentation: bool, + deadline: Instant, +) -> DescribeConfigsResult { + if !resource.unknown_tagged_fields.is_empty() { + return resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(static_text("unknown tagged fields are not supported")), + Vec::new(), + ); + } + if resource.resource_type != RESOURCE_TYPE_TOPIC { + return resource_error( + resource, + ERROR_INVALID_REQUEST, + Some(static_text(ONLY_TOPIC_RESOURCES)), + Vec::new(), + ); + } + 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))), + Vec::new(), + ); + } + + if Instant::now() >= deadline { + tracing::warn!( + resource = name, + "DescribeConfigs deadline passed before the topic lookup" + ); + return resource_error(resource, ERROR_REQUEST_TIMED_OUT, None, Vec::new()); + } + let topic = match tokio::time::timeout_at(deadline, bridge.get_kafka_topic(name)).await { + Ok(Ok(Some(topic))) => topic, + Ok(Ok(None)) => { + return resource_error(resource, ERROR_UNKNOWN_TOPIC_OR_PARTITION, None, Vec::new()); + } + Ok(Err(error)) => { + let (code, message) = bridge_failure(&error, "DescribeConfigs", "reading"); + return resource_error(resource, code, message, Vec::new()); + } + Err(_elapsed) => { + tracing::warn!( + resource = name, + "DescribeConfigs: this resource's bridge work exceeded the request deadline; \ + answering retriable instead of blocking further" + ); + return resource_error(resource, ERROR_REQUEST_TIMED_OUT, None, Vec::new()); + } + }; + + let (stream, iggy_topic) = bridge.topic_identity(name); + let remembered = state.remembered_retention_synonyms(&IggyTopicKey::new(stream, iggy_topic)); + let (configs, unknown_key, unrepresentable) = described_entries( + &topic, + resource.configuration_keys.as_deref(), + include_synonyms, + include_documentation, + remembered, + ); + + if let Some(unknown_key) = unknown_key { Review Comment: warning: An unmodeled requested key fails the whole resource with `INVALID_CONFIG` (40), and clients discard configs on error, so such a request loses `retention.ms` too. Answer `ERROR_NONE` and omit the unknown key, as a real broker does. ########## gateways/kafka/tests/version_firewall_tests.rs: ########## @@ -101,7 +103,7 @@ fn is_supported_version_matches_scope_table() { /// /// Relies on `SUPPORTED_RANGES` (src) and `SCOPED_API_KEYS` (test) sharing declaration order /// (Produce, Fetch, `ListOffsets`, Metadata, `ApiVersions`, `CreateTopics`, `FindCoordinator`, -/// `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`) - `supported_ranges_table_has_twelve_entries` plus +/// `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`) - `supported_ranges_table_has_fourteen_entries` plus Review Comment: nit: The comment says `SUPPORTED_RANGES` and `SCOPED_API_KEYS` share declaration order, but `api_versions.rs` sorts the wire rows by `api_key` and the scope table is the ascending one. Rewrite it to say that the test relies on that sort. ########## 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: simplification: The `unit_ms == 1` branch calls `parse_retention_ms(value)`, which is exactly `parse_scaled_retention(value, 1)`. Delete the branch and call `parse_scaled_retention(value, unit_ms)` for every unit. -- 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]
