You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ra...@apache.org on 2026/10/04 03:43:50 UTC
(iggy) 01/01: feat(gateways): add Kafka DescribeConfigs and AlterConfigs Kafka admin clients need to read and set topic retention. DescribeConfigs reports retention.ms from Iggy message expiry and a read-only cleanup.policy. AlterConfigs persists only retention.ms, capped so segment expiry cannot wrap.
This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a commit to branch feat/gateways-3546-describe-alter-configs
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 80b156538f2097bcd1c107f224f82e1a1e9e48f4
Author: ryerraguntla <ra...@gmail.com>
AuthorDate: Sat Oct 3 23:43:11 2026 -0400
feat(gateways): add Kafka DescribeConfigs and AlterConfigs Kafka admin clients need to read and set topic retention. DescribeConfigs
reports retention.ms from Iggy message expiry and a read-only cleanup.policy.
AlterConfigs persists only retention.ms, capped so segment expiry cannot wrap.
---
gateways/kafka/docs/CONFIGS.md | 67 +++
.../kafka/src/protocol/handlers/alter_configs.rs | 359 ++++++++++++++
.../src/protocol/handlers/describe_configs.rs | 439 +++++++++++++++++
.../kafka/src/protocol/handlers/topic_config.rs | 348 ++++++++++++++
gateways/kafka/tests/configs_handler_tests.rs | 113 +++++
gateways/kafka/tests/configs_java_client_tests.rs | 265 +++++++++++
gateways/kafka/tests/configs_real_bridge_tests.rs | 521 +++++++++++++++++++++
gateways/kafka/tests/java/ConfigsProbe.java | 129 +++++
8 files changed, 2241 insertions(+)
diff --git a/gateways/kafka/docs/CONFIGS.md b/gateways/kafka/docs/CONFIGS.md
new file mode 100644
index 000000000..b734ca0a1
--- /dev/null
+++ b/gateways/kafka/docs/CONFIGS.md
@@ -0,0 +1,67 @@
+# Topic configuration
+
+`DescribeConfigs` (32) and `AlterConfigs` (33) apply to Kafka topic resources only
+(resource type 2). Only the deprecated `Admin.alterConfigs` (API 33) writes
+`retention.ms`. `IncrementalAlterConfigs` (44) stays unimplemented: it is not
+advertised and closes the connection like every other unlisted key.
+
+With the bridge off, both APIs answer `NOT_CONTROLLER` (41) for each resource and
+do not read or write Iggy.
+
+## What a describe returns
+
+A null or empty `configuration_keys` list returns both keys below. A non-empty
+list returns only the requested names, in request order. A name this gateway does
+not know is still returned, and that resource's error code is `INVALID_CONFIG`
+(40).
+
+| Key | Value | Read only | Source |
+| --- | ----- | --------- | ------ |
+| `retention.ms` | Sealed-segment lifetime from Iggy `message_expiry`, or `-1` when those segments never expire | no | `DEFAULT_CONFIG` (5) for the unset never-expire default; `DYNAMIC_TOPIC_CONFIG` (1) when a client set it or a stored duration cannot be shown as milliseconds |
+| `cleanup.policy` | `delete` | yes | `DEFAULT_CONFIG` (5) |
+
+`is_sensitive` is false for both. `include_synonyms` is honored and the synonym
+list is empty either way. From v3, `include_documentation` is honored: true
+attaches one sentence per known key, false sends no documentation.
+
+Iggy stores `message_expiry` as `IggyExpiry`. A duration is `IggyDuration`, and
+that type counts microseconds. Retention applies to sealed segments only. Iggy
+does not time-roll, so the active segment does not expire. Describe divides by
+1000 only when the stored count is a positive whole number of milliseconds.
+Anything else, including 0 microseconds, is `INVALID_CONFIG` (40) with a null
+value and source `DYNAMIC_TOPIC_CONFIG` (1). That expiry was stored; it is not
+the never-expire default. The gateway does not round.
+
+## What an alter stores
+
+Only `retention.ms` is written, through `update_topic`'s `message_expiry`.
+
+- `-1` stores `IggyExpiry::NeverExpire`. The server treats `0`
+ (`IggyExpiry::ServerDefault`) as "leave the current expiry", so `-1` is not
+ sent as 0. A following describe reports `-1` with source
+ `DYNAMIC_TOPIC_CONFIG`.
+- A positive millisecond count is multiplied by 1000 and stored as
+ `IggyExpiry::ExpireDuration`. The text must be the canonical decimal of that
+ integer (`1`, not `0001` or `+1`). The duration must be at most `u32::MAX`
+ seconds, the same limit `IggyExpiry::from_str` enforces. A longer value is
+ `INVALID_CONFIG` (40) and is not stored. A following describe reports the same
+ decimal with source `DYNAMIC_TOPIC_CONFIG`.
+- `0` and every other value are `INVALID_CONFIG` (40).
+
+`cleanup.policy` and any other key are `INVALID_CONFIG` (40). They are not
+stored. Kafka has no `NOT_CONFIGURABLE` code. One bad key fails that resource
+and writes nothing for it. Other resources in the same request are still
+processed.
+
+`validate_only` runs the same checks, including the topic lookup, and does not
+call `update_topic`.
+
+## Other resource errors
+
+| Condition | Code | Notes |
+| --------- | ---- | ----- |
+| Resource type other than topic | `INVALID_REQUEST` (42) | Message is `only topic resources are supported`. Other resources in the batch still run |
+| Topic name fails the same rules as CreateTopics | `INVALID_TOPIC_EXCEPTION` (17) | The message is the validation reason and does not repeat the topic name |
+| Topic does not exist | `UNKNOWN_TOPIC_OR_PARTITION` (3) | Checked before config keys, so a missing topic is not `INVALID_CONFIG` |
+| More than 100 distinct topic names | `POLICY_VIOLATION` (44) | Every resource in the request. Duplicate names count once and are not an error by themselves |
+| Non-empty tagged fields | `INVALID_REQUEST` (42) | Request-level tags fail every resource. Per-resource or per-entry tags fail that resource |
diff --git a/gateways/kafka/src/protocol/handlers/alter_configs.rs b/gateways/kafka/src/protocol/handlers/alter_configs.rs
new file mode 100644
index 000000000..26154fb75
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/alter_configs.rs
@@ -0,0 +1,359 @@
+// 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 std::collections::HashMap;
+
+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::{
+ ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC, UNKNOWN_TAGGED_FIELDS, exceeds_topic_cap,
+ plan_retention_update, 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",
+ );
+ }
+ };
+
+ if !req.unknown_tagged_fields.is_empty() {
+ return respond_or_close(
+ encode_message(
+ &response(reject_tagged_resources(&req.resources)),
+ api_version,
+ 256,
+ ),
+ "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(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(
+ bridge: &IggyBridge,
+ req: &AlterConfigsRequest,
+ deadline: Instant,
+) -> Vec<AlterConfigsResourceResponse> {
+ let topic_names = req
+ .resources
+ .iter()
+ .filter(|resource| resource.resource_type == RESOURCE_TYPE_TOPIC)
+ .map(|resource| resource.resource_name.as_str());
+ if exceeds_topic_cap(topic_names) {
+ 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();
+ }
+
+ let mut seen: HashMap<String, AlterConfigsResourceResponse> = HashMap::new();
+ 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
+ && let Some(prior) = seen.get(resource.resource_name.as_str())
+ {
+ responses.push(prior.clone());
+ 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"
+ );
+ }
+ let timed_out = resource_error(resource, ERROR_REQUEST_TIMED_OUT, None);
+ remember_topic(&mut seen, resource, &timed_out);
+ responses.push(timed_out);
+ continue;
+ }
+
+ let result = alter_one(bridge, resource, req.validate_only, deadline).await;
+ if Instant::now() >= deadline {
+ deadline_exceeded = true;
+ }
+ remember_topic(&mut seen, resource, &result);
+ responses.push(result);
+ }
+ responses
+}
+
+fn remember_topic(
+ seen: &mut HashMap<String, AlterConfigsResourceResponse>,
+ resource: &AlterConfigsResource,
+ result: &AlterConfigsResourceResponse,
+) {
+ if resource.resource_type == RESOURCE_TYPE_TOPIC {
+ seen.insert(resource.resource_name.as_str().to_string(), result.clone());
+ }
+}
+
+async fn alter_one(
+ 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)),
+ );
+ }
+ 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 bridge.get_kafka_topic(name).await {
+ Ok(Some(_topic)) => {}
+ Ok(None) => {
+ return resource_error(resource, ERROR_UNKNOWN_TOPIC_OR_PARTITION, None);
+ }
+ Err(error) => {
+ let (code, message) = bridge_failure(&error);
+ return resource_error(resource, code, message);
+ }
+ }
+
+ let expiry = match plan_retention_update(resource.configs.iter().map(|config| {
+ (
+ config.name.as_str(),
+ config.value.as_ref().map(StrBytes::as_str),
+ )
+ })) {
+ Ok(expiry) => expiry,
+ Err(fault) => {
+ return resource_error(
+ resource,
+ ERROR_INVALID_CONFIG,
+ Some(StrBytes::from(fault.message())),
+ );
+ }
+ };
+
+ // `validate_only` already performed the existence read and the same key checks.
+ // It does not call `update_topic`.
+ let Some(expiry) = expiry else {
+ return resource_error(resource, ERROR_NONE, None);
+ };
+ 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
+ // `timeout_at`. Dropping that future does not abort the SDK task, so
+ // `UpdateTopic` can commit after this handler has already answered
+ // REQUEST_TIMED_OUT. A short retention then deletes sealed segments.
+ if Instant::now() >= deadline {
+ tracing::warn!(
+ resource = name,
+ "AlterConfigs deadline passed before update_topic; not starting the write"
+ );
+ return resource_error(resource, ERROR_REQUEST_TIMED_OUT, None);
+ }
+ match bridge.update_kafka_topic_message_expiry(name, expiry).await {
+ Ok(()) | Err(BridgeError::Iggy(IggyError::RequestAlreadyApplied)) => {
+ resource_error(resource, ERROR_NONE, None)
+ }
+ Err(error) => {
+ let (code, message) = bridge_failure(&error);
+ resource_error(resource, code, message)
+ }
+ }
+}
+
+fn reject_tagged_resources(
+ resources: &[AlterConfigsResource],
+) -> Vec<AlterConfigsResourceResponse> {
+ if resources.is_empty() {
+ return vec![
+ AlterConfigsResourceResponse::default()
+ .with_error_code(ERROR_INVALID_REQUEST)
+ .with_error_message(Some(static_text(UNKNOWN_TAGGED_FIELDS)))
+ .with_resource_type(0)
+ .with_resource_name(StrBytes::from_static_str("")),
+ ];
+ }
+ resources
+ .iter()
+ .map(|resource| {
+ resource_error(
+ resource,
+ ERROR_INVALID_REQUEST,
+ Some(static_text(UNKNOWN_TAGGED_FIELDS)),
+ )
+ })
+ .collect()
+}
+
+fn resource_error(
+ resource: &AlterConfigsResource,
+ error_code: i16,
+ error_message: Option<StrBytes>,
+) -> AlterConfigsResourceResponse {
+ AlterConfigsResourceResponse::default()
+ .with_error_code(error_code)
+ .with_error_message(error_message)
+ .with_resource_type(resource.resource_type)
+ .with_resource_name(resource.resource_name.clone())
+}
+
+fn name_reason(error: &BridgeError) -> String {
+ match error {
+ BridgeError::InvalidKafkaTopicName { reason, .. } => reason.clone(),
+ _ => "invalid topic name".to_string(),
+ }
+}
+
+fn bridge_failure(error: &BridgeError) -> (i16, Option<StrBytes>) {
+ let code = error.to_kafka_error_code();
+ match error {
+ BridgeError::InvalidKafkaTopicName { reason, .. } => {
+ tracing::debug!(reason, "AlterConfigs rejected an invalid topic name");
+ (code, Some(StrBytes::from(reason.clone())))
+ }
+ BridgeError::Timeout | BridgeError::SendLost(_) => {
+ tracing::warn!(%error, "AlterConfigs exceeded the bridge deadline");
+ (code, None)
+ }
+ other => {
+ tracing::error!(%other, "AlterConfigs failed while updating topic configuration");
+ (
+ code,
+ Some(static_text("internal error updating topic configuration")),
+ )
+ }
+ }
+}
+
+const fn static_text(message: &'static str) -> StrBytes {
+ StrBytes::from_static_str(message)
+}
diff --git a/gateways/kafka/src/protocol/handlers/describe_configs.rs b/gateways/kafka/src/protocol/handlers/describe_configs.rs
new file mode 100644
index 000000000..530b305a4
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/describe_configs.rs
@@ -0,0 +1,439 @@
+// 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 std::collections::HashMap;
+
+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::topic_map::validate_kafka_topic_name;
+use crate::bridge::{BridgeError, IggyBridge};
+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, ListedKey, ONLY_TOPIC_RESOURCES, RESOURCE_TYPE_TOPIC,
+ RETENTION_DOC, RETENTION_MS, UNKNOWN_TAGGED_FIELDS, exceeds_topic_cap, listed_keys,
+ message_expiry_is_explicit, retention_ms, 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",
+ );
+ }
+ };
+
+ if !req.unknown_tagged_fields.is_empty() {
+ return respond_or_close(
+ encode_message(
+ &response(reject_tagged_resources(&req.resources)),
+ api_version,
+ 256,
+ ),
+ "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(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(
+ bridge: &IggyBridge,
+ req: &DescribeConfigsRequest,
+ deadline: Instant,
+) -> Vec<DescribeConfigsResult> {
+ let topic_names = req
+ .resources
+ .iter()
+ .filter(|resource| resource.resource_type == RESOURCE_TYPE_TOPIC)
+ .map(|resource| resource.resource_name.as_str());
+ if exceeds_topic_cap(topic_names) {
+ 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();
+ }
+
+ // `get_kafka_topics` lists partition counts, not `message_expiry`, so each distinct
+ // topic is one `get_kafka_topic`. Duplicates reuse the first result and do not call again.
+ let mut seen: HashMap<String, DescribeConfigsResult> = HashMap::new();
+ 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
+ && let Some(prior) = seen.get(resource.resource_name.as_str())
+ {
+ results.push(prior.clone());
+ 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"
+ );
+ }
+ let timed_out = resource_error(resource, ERROR_REQUEST_TIMED_OUT, None, Vec::new());
+ remember_topic(&mut seen, resource, &timed_out);
+ results.push(timed_out);
+ continue;
+ }
+
+ let result = describe_one(
+ bridge,
+ resource,
+ req.include_synonyms,
+ req.include_documentation,
+ deadline,
+ )
+ .await;
+ if Instant::now() >= deadline {
+ deadline_exceeded = true;
+ }
+ remember_topic(&mut seen, resource, &result);
+ results.push(result);
+ }
+ results
+}
+
+fn remember_topic(
+ seen: &mut HashMap<String, DescribeConfigsResult>,
+ resource: &DescribeConfigsResource,
+ result: &DescribeConfigsResult,
+) {
+ if resource.resource_type == RESOURCE_TYPE_TOPIC {
+ seen.insert(resource.resource_name.as_str().to_string(), result.clone());
+ }
+}
+
+async fn describe_one(
+ 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)),
+ 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 bridge.get_kafka_topic(name).await {
+ Ok(Some(topic)) => topic,
+ Ok(None) => {
+ return resource_error(resource, ERROR_UNKNOWN_TOPIC_OR_PARTITION, None, Vec::new());
+ }
+ Err(error) => {
+ let (code, message) = bridge_failure(&error, "reading");
+ return resource_error(resource, code, message, Vec::new());
+ }
+ };
+
+ let (configs, unknown, unrepresentable) = described_entries(
+ &topic,
+ resource.configuration_keys.as_deref(),
+ include_synonyms,
+ include_documentation,
+ );
+
+ if unknown || unrepresentable {
+ let message = if unrepresentable {
+ "stored message expiry is not a whole number of milliseconds"
+ } else {
+ "unknown config key"
+ };
+ return resource_error(
+ resource,
+ ERROR_INVALID_CONFIG,
+ Some(static_text(message)),
+ configs,
+ );
+ }
+ resource_error(resource, ERROR_NONE, None, configs)
+}
+
+#[must_use]
+fn described_entries(
+ topic: &TopicDetails,
+ configuration_keys: Option<&[StrBytes]>,
+ include_synonyms: bool,
+ include_documentation: bool,
+) -> (Vec<DescribeConfigsResourceResult>, bool, bool) {
+ let explicit = message_expiry_is_explicit(&topic.options);
+ let retention = retention_ms(topic.message_expiry, explicit);
+ let keys = listed_keys(configuration_keys);
+ let mut unknown = false;
+ let mut unrepresentable = false;
+ let mut configs = Vec::with_capacity(keys.len());
+ for key in keys {
+ match key {
+ ListedKey::Retention => {
+ let (value, source) = retention.as_ref().map_or_else(
+ |()| {
+ // A stored duration that cannot be shown as retention.ms is still the
+ // topic's expiry, not the never-expire default.
+ unrepresentable = true;
+ (None, CONFIG_SOURCE_TOPIC)
+ },
+ |retention| (Some(retention.value.as_str()), retention.source),
+ );
+ configs.push(config_entry(
+ RETENTION_MS,
+ value,
+ false,
+ source,
+ CONFIG_TYPE_LONG,
+ include_synonyms,
+ include_documentation.then_some(RETENTION_DOC),
+ ));
+ }
+ ListedKey::Cleanup => configs.push(config_entry(
+ CLEANUP_POLICY,
+ Some(CLEANUP_POLICY_VALUE),
+ true,
+ CONFIG_SOURCE_DEFAULT,
+ CONFIG_TYPE_STRING,
+ include_synonyms,
+ include_documentation.then_some(CLEANUP_DOC),
+ )),
+ ListedKey::Unknown(name) => {
+ unknown = true;
+ configs.push(config_entry(
+ &name,
+ None,
+ false,
+ CONFIG_SOURCE_DEFAULT,
+ 0,
+ include_synonyms,
+ None,
+ ));
+ }
+ }
+ }
+ (configs, unknown, unrepresentable)
+}
+
+fn config_entry(
+ name: &str,
+ value: Option<&str>,
+ read_only: bool,
+ source: i8,
+ config_type: i8,
+ include_synonyms: bool,
+ documentation: Option<&str>,
+) -> DescribeConfigsResourceResult {
+ DescribeConfigsResourceResult::default()
+ .with_name(StrBytes::from(name.to_string()))
+ .with_value(value.map(|value| StrBytes::from(value.to_string())))
+ .with_read_only(read_only)
+ .with_config_source(source)
+ .with_is_sensitive(false)
+ .with_synonyms(synonyms(include_synonyms))
+ .with_config_type(config_type)
+ .with_documentation(documentation.map(|text| StrBytes::from(text.to_string())))
+}
+
+/// Neither known key has a synonym. `include_synonyms` asks for the list and receives none.
+const fn synonyms(_include_synonyms: bool) -> Vec<DescribeConfigsSynonym> {
+ Vec::new()
+}
+
+fn reject_tagged_resources(resources: &[DescribeConfigsResource]) -> Vec<DescribeConfigsResult> {
+ if resources.is_empty() {
+ return vec![
+ DescribeConfigsResult::default()
+ .with_error_code(ERROR_INVALID_REQUEST)
+ .with_error_message(Some(static_text(UNKNOWN_TAGGED_FIELDS)))
+ .with_resource_type(0)
+ .with_resource_name(StrBytes::from_static_str(""))
+ .with_configs(Vec::new()),
+ ];
+ }
+ resources
+ .iter()
+ .map(|resource| {
+ resource_error(
+ resource,
+ ERROR_INVALID_REQUEST,
+ Some(static_text(UNKNOWN_TAGGED_FIELDS)),
+ Vec::new(),
+ )
+ })
+ .collect()
+}
+
+fn resource_error(
+ resource: &DescribeConfigsResource,
+ error_code: i16,
+ error_message: Option<StrBytes>,
+ configs: Vec<DescribeConfigsResourceResult>,
+) -> DescribeConfigsResult {
+ DescribeConfigsResult::default()
+ .with_error_code(error_code)
+ .with_error_message(error_message)
+ .with_resource_type(resource.resource_type)
+ .with_resource_name(resource.resource_name.clone())
+ .with_configs(configs)
+}
+
+fn name_reason(error: &BridgeError) -> String {
+ match error {
+ BridgeError::InvalidKafkaTopicName { reason, .. } => reason.clone(),
+ _ => "invalid topic name".to_string(),
+ }
+}
+
+fn bridge_failure(error: &BridgeError, action: &str) -> (i16, Option<StrBytes>) {
+ let code = error.to_kafka_error_code();
+ match error {
+ BridgeError::InvalidKafkaTopicName { reason, .. } => {
+ tracing::debug!(reason, "DescribeConfigs rejected an invalid topic name");
+ (code, Some(StrBytes::from(reason.clone())))
+ }
+ BridgeError::Timeout | BridgeError::SendLost(_) => {
+ tracing::warn!(%error, "DescribeConfigs {action} exceeded the bridge deadline");
+ (code, None)
+ }
+ other => {
+ tracing::error!(%other, "DescribeConfigs failed while {action} topic configuration");
+ (
+ code,
+ Some(static_text("internal error reading topic configuration")),
+ )
+ }
+ }
+}
+
+const fn static_text(message: &'static str) -> StrBytes {
+ StrBytes::from_static_str(message)
+}
diff --git a/gateways/kafka/src/protocol/handlers/topic_config.rs b/gateways/kafka/src/protocol/handlers/topic_config.rs
new file mode 100644
index 000000000..876092282
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/topic_config.rs
@@ -0,0 +1,348 @@
+// 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. A positive count is capped at `u32::MAX` seconds, matching
+//! [`IggyExpiry::from_str`].
+
+use std::collections::HashSet;
+use std::str::FromStr;
+
+use iggy::prelude::{HeaderKey, IggyDuration, IggyExpiry, ResourceOptions, topic_option_keys};
+use kafka_protocol::protocol::StrBytes;
+
+/// 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 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 UNKNOWN_TAGGED_FIELDS: &str = "unknown tagged fields are not 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 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. 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,
+ 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,
+}
+
+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 must be -1 or a positive number of milliseconds at most {} seconds",
+ u32::MAX
+ ),
+ }
+ }
+}
+
+/// `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(());
+ }
+ let millis = i64::try_from(micros / 1_000).map_err(|_| ())?;
+ if millis <= 0 {
+ return Err(());
+ }
+ Ok(RetentionMs {
+ value: millis.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> {
+ let Ok(millis) = value.parse::<i64>() else {
+ return Err(ConfigFault::InvalidRetention);
+ };
+ if millis.to_string() != value {
+ return Err(ConfigFault::InvalidRetention);
+ }
+ if millis == -1 {
+ return Ok(IggyExpiry::NeverExpire);
+ }
+ if millis <= 0 {
+ return Err(ConfigFault::InvalidRetention);
+ }
+ let millis = u64::try_from(millis).map_err(|_| ConfigFault::InvalidRetention)?;
+ // `IggyExpiry::from_str` rejects `as_secs() > u32::MAX`. A longer duration makes
+ // `Segment::is_expired` wrap `max_timestamp + micros` (and panic in debug), so a
+ // sealed segment looks expired and is deleted.
+ 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"
+ )
+}
+
+/// Keys to return for one describe resource.
+///
+/// `None` and an empty list both mean both known keys. 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,
+ CLEANUP_POLICY => ListedKey::Cleanup,
+ other => ListedKey::Unknown(other.to_string()),
+ })
+ .collect()
+}
+
+/// The expiry to store, or `None` when the resource names no `retention.ms`.
+///
+/// One invalid or repeated key fails the whole resource. Nothing is selected
+/// for storage in that case.
+///
+/// # 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<IggyExpiry>, ConfigFault> {
+ let mut seen = HashSet::new();
+ let mut expiry = None;
+ for (name, value) in configs {
+ if !seen.insert(name) {
+ return Err(ConfigFault::Repeated(name.to_string()));
+ }
+ match name {
+ RETENTION_MS => {
+ let Some(value) = value else {
+ return Err(ConfigFault::InvalidRetention);
+ };
+ expiry = Some(parse_retention_ms(value)?);
+ }
+ CLEANUP_POLICY => {
+ return Err(ConfigFault::ReadOnly("cleanup.policy is read-only"));
+ }
+ other => return Err(ConfigFault::UnknownKey(other.to_string())),
+ }
+ }
+ Ok(expiry)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn retention_ms_round_trips_a_positive_value_and_never_expire() {
+ let expiry = parse_retention_ms("1500").expect("1500 ms");
+ let described = retention_ms(expiry, true).expect("representable");
+ assert_eq!(described.value, "1500");
+ assert_eq!(described.source, CONFIG_SOURCE_TOPIC);
+ assert_eq!(
+ u64::from(expiry),
+ 1_500_000,
+ "IggyDuration stores microseconds"
+ );
+
+ let never = parse_retention_ms("-1").expect("never expire");
+ assert_eq!(never, IggyExpiry::NeverExpire);
+ let described = retention_ms(never, true).expect("never expire");
+ assert_eq!(described.value, "-1");
+ assert_eq!(described.source, CONFIG_SOURCE_TOPIC);
+ let unset = retention_ms(IggyExpiry::NeverExpire, false).expect("default");
+ assert_eq!(unset.value, "-1");
+ assert_eq!(unset.source, CONFIG_SOURCE_DEFAULT);
+ }
+
+ #[test]
+ fn retention_ms_rejects_zero_noncanonical_and_unrepresentable_values() {
+ assert!(parse_retention_ms("0").is_err());
+ assert!(parse_retention_ms("-2").is_err());
+ assert!(parse_retention_ms("0001").is_err());
+ assert!(parse_retention_ms("+1").is_err());
+ assert!(parse_retention_ms("1.5").is_err());
+ assert!(parse_retention_ms("").is_err());
+ assert!(parse_retention_ms(&i64::MAX.to_string()).is_err());
+ let max_secs_ms = u64::from(u32::MAX) * 1_000;
+ let at_cap = parse_retention_ms(&max_secs_ms.to_string()).expect("u32::MAX seconds");
+ assert_eq!(u64::from(at_cap), max_secs_ms * 1_000);
+ let with_remainder = max_secs_ms + 999;
+ assert!(
+ parse_retention_ms(&with_remainder.to_string()).is_ok(),
+ "a remainder under one second stays within IggyExpiry::from_str"
+ );
+ assert!(parse_retention_ms(&(max_secs_ms + 1_000).to_string()).is_err());
+ assert!(parse_retention_ms(&(u64::MAX / 1_000).to_string()).is_err());
+
+ let sub_ms = IggyExpiry::ExpireDuration(IggyDuration::from(1_500_u64));
+ assert!(retention_ms(sub_ms, true).is_err());
+ }
+
+ #[test]
+ fn listed_keys_defaults_to_both_and_keeps_an_unknown_name() {
+ assert_eq!(
+ listed_keys(None),
+ vec![ListedKey::Retention, ListedKey::Cleanup]
+ );
+ assert_eq!(
+ listed_keys(Some(&[])),
+ vec![ListedKey::Retention, ListedKey::Cleanup]
+ );
+ let asked = [
+ StrBytes::from_static_str(CLEANUP_POLICY),
+ StrBytes::from_static_str("no.such"),
+ ];
+ assert_eq!(
+ listed_keys(Some(&asked)),
+ vec![
+ ListedKey::Cleanup,
+ ListedKey::Unknown("no.such".to_string())
+ ]
+ );
+ }
+
+ #[test]
+ fn one_bad_key_rejects_the_whole_alter_plan() {
+ let err = plan_retention_update([(RETENTION_MS, Some("5000")), ("no.such", Some("1"))])
+ .expect_err("unknown key");
+ assert_eq!(err, ConfigFault::UnknownKey("no.such".to_string()));
+
+ let err = plan_retention_update([(CLEANUP_POLICY, Some("delete"))]).expect_err("read only");
+ assert!(matches!(err, ConfigFault::ReadOnly(_)));
+
+ assert_eq!(
+ plan_retention_update([(RETENTION_MS, Some("-1"))]).expect("clear"),
+ Some(IggyExpiry::NeverExpire)
+ );
+ assert_eq!(
+ plan_retention_update([] as [(&str, Option<&str>); 0]).expect("empty"),
+ None
+ );
+ }
+}
diff --git a/gateways/kafka/tests/configs_handler_tests.rs b/gateways/kafka/tests/configs_handler_tests.rs
new file mode 100644
index 000000000..ede44539b
--- /dev/null
+++ b/gateways/kafka/tests/configs_handler_tests.rs
@@ -0,0 +1,113 @@
+// 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` and `AlterConfigs` with the bridge off.
+
+use bytes::{Bytes, BytesMut};
+use kafka_protocol::messages::alter_configs_request::{
+ AlterConfigsRequest, AlterConfigsResource, AlterableConfig,
+};
+use kafka_protocol::messages::alter_configs_response::AlterConfigsResponse;
+use kafka_protocol::messages::describe_configs_request::{
+ DescribeConfigsRequest, DescribeConfigsResource,
+};
+use kafka_protocol::messages::describe_configs_response::DescribeConfigsResponse;
+use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::api::{BrokerAdvertise, ERROR_NOT_CONTROLLER, GatewayState};
+use iggy_gateway_kafka::protocol::handlers::{alter_configs, describe_configs};
+
+const DESCRIBE_VERSION: i16 = 4;
+const ALTER_VERSION: i16 = 2;
+
+fn state_without_bridge() -> GatewayState {
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ None,
+ 8 * 1024 * 1024,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+}
+
+fn encode<M: Encodable>(message: &M, version: i16) -> Bytes {
+ let mut buf = BytesMut::new();
+ message.encode(&mut buf, version).expect("encode request");
+ buf.freeze()
+}
+
+fn decode<M: Decodable>(body: Bytes, version: i16) -> M {
+ let mut buf = body;
+ M::decode(&mut buf, version).expect("decode response")
+}
+
+#[tokio::test]
+async fn describe_configs_without_a_bridge_returns_not_controller() {
+ let request = DescribeConfigsRequest::default()
+ .with_resources(vec![
+ DescribeConfigsResource::default()
+ .with_resource_type(2)
+ .with_resource_name(StrBytes::from_static_str("orders"))
+ .with_configuration_keys(None),
+ ])
+ .with_include_synonyms(false)
+ .with_include_documentation(false);
+ let outcome = describe_configs::handle(
+ &state_without_bridge(),
+ DESCRIBE_VERSION,
+ encode(&request, DESCRIBE_VERSION),
+ )
+ .await;
+ let response: DescribeConfigsResponse = decode(
+ outcome.expect_response("DescribeConfigs answers"),
+ DESCRIBE_VERSION,
+ );
+ assert_eq!(response.results.len(), 1);
+ assert_eq!(response.results[0].error_code, ERROR_NOT_CONTROLLER);
+ assert!(response.results[0].configs.is_empty());
+}
+
+#[tokio::test]
+async fn alter_configs_without_a_bridge_returns_not_controller() {
+ let request = AlterConfigsRequest::default()
+ .with_resources(vec![
+ AlterConfigsResource::default()
+ .with_resource_type(2)
+ .with_resource_name(StrBytes::from_static_str("orders"))
+ .with_configs(vec![
+ AlterableConfig::default()
+ .with_name(StrBytes::from_static_str("retention.ms"))
+ .with_value(Some(StrBytes::from_static_str("1000"))),
+ ]),
+ ])
+ .with_validate_only(false);
+ let outcome = alter_configs::handle(
+ &state_without_bridge(),
+ ALTER_VERSION,
+ encode(&request, ALTER_VERSION),
+ )
+ .await;
+ let response: AlterConfigsResponse = decode(
+ outcome.expect_response("AlterConfigs answers"),
+ ALTER_VERSION,
+ );
+ assert_eq!(response.responses.len(), 1);
+ assert_eq!(response.responses[0].error_code, ERROR_NOT_CONTROLLER);
+}
diff --git a/gateways/kafka/tests/configs_java_client_tests.rs b/gateways/kafka/tests/configs_java_client_tests.rs
new file mode 100644
index 000000000..4dc590508
--- /dev/null
+++ b/gateways/kafka/tests/configs_java_client_tests.rs
@@ -0,0 +1,265 @@
+// 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);
+
+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!(
+ output.status.success(),
+ "download {url} failed: {}{}",
+ output.status,
+ String::from_utf8_lossy(&output.stderr)
+ );
+ }
+}
+
+fn classpath(jars: &Path, classes: &Path) -> String {
+ let mut parts = vec![classes.display().to_string()];
+ for (name, _) in JARS {
+ parts.push(jars.join(name).display().to_string());
+ }
+ parts.join(":")
+}
+
+async fn spawn_gateway(server: &TestServer) -> SocketAddr {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge connects");
+ bridge
+ .ensure_stream_and_topic(TOPIC, 1)
+ .await
+ .expect("create topic");
+ let listener = bind_listener("127.0.0.1:0").expect("bind");
+ let addr = listener.local_addr().expect("local addr");
+ let gateway = KafkaGateway::new(GatewayConfig {
+ bind_addr: addr.to_string(),
+ ..GatewayConfig::default()
+ })
+ .with_bridge(Some(Arc::new(bridge)));
+ let (shutdown, receiver) = broadcast::channel::<()>(1);
+ tokio::spawn(async move {
+ let _ = gateway.run(listener, receiver).await;
+ });
+ std::mem::forget(shutdown);
+ addr
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn kafka_39_admin_client_describes_and_alters_topic_configs() {
+ let java = output_bounded(Command::new("java").arg("-version"), "java -version");
+ assert!(
+ java.status.success(),
+ "java -version failed: {}",
+ String::from_utf8_lossy(&java.stderr)
+ );
+
+ let jars = jar_dir();
+ ensure_jars(&jars);
+ let classes = tempfile::tempdir().expect("class dir");
+ let source = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/java/ConfigsProbe.java");
+ let javac = output_bounded(
+ Command::new("javac")
+ .arg("--release")
+ .arg("17")
+ .arg("-cp")
+ .arg(classpath(&jars, classes.path()))
+ .arg("-d")
+ .arg(classes.path())
+ .arg(&source),
+ "javac",
+ );
+ assert!(
+ javac.status.success(),
+ "javac failed\n{}",
+ String::from_utf8_lossy(&javac.stderr)
+ );
+
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let addr = spawn_gateway(&server).await;
+
+ let output = output_bounded(
+ Command::new("java")
+ .arg("-cp")
+ .arg(classpath(&jars, classes.path()))
+ .arg("ConfigsProbe")
+ .arg(addr.to_string())
+ .arg(TOPIC),
+ "java ConfigsProbe",
+ );
+ let text = format!(
+ "{}{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ );
+ println!("{text}");
+ assert!(
+ output.status.success(),
+ "ConfigsProbe failed ({:?})\n{text}",
+ output.status.code()
+ );
+ assert!(text.contains("CONFIGS_PROBE_OK"), "{text}");
+ assert!(
+ text.contains("describe_batch_50_elapsed_ms="),
+ "missing timing line\n{text}"
+ );
+ assert!(
+ text.contains("retention.ms=-1 source=DEFAULT_CONFIG"),
+ "{text}"
+ );
+ assert!(
+ text.contains("cleanup.policy=delete read_only=true source=DEFAULT_CONFIG"),
+ "{text}"
+ );
+ assert!(
+ text.contains("unknown_key_rejected=InvalidConfigurationException"),
+ "{text}"
+ );
+ assert!(text.contains("validate_only_retention.ms=-1"), "{text}");
+ assert!(
+ text.contains("altered_retention.ms=8000 source=DYNAMIC_TOPIC_CONFIG"),
+ "{text}"
+ );
+}
diff --git a/gateways/kafka/tests/configs_real_bridge_tests.rs b/gateways/kafka/tests/configs_real_bridge_tests.rs
new file mode 100644
index 000000000..933a85a9f
--- /dev/null
+++ b/gateways/kafka/tests/configs_real_bridge_tests.rs
@@ -0,0 +1,521 @@
+// 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` and `AlterConfigs` against a real `iggy-server`.
+
+use std::sync::Arc;
+
+use bytes::{Bytes, BytesMut};
+use kafka_protocol::messages::alter_configs_request::{
+ AlterConfigsRequest, AlterConfigsResource, AlterableConfig,
+};
+use kafka_protocol::messages::alter_configs_response::AlterConfigsResponse;
+use kafka_protocol::messages::describe_configs_request::{
+ DescribeConfigsRequest, DescribeConfigsResource,
+};
+use kafka_protocol::messages::describe_configs_response::{
+ DescribeConfigsResourceResult, DescribeConfigsResponse,
+};
+use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
+use serial_test::serial;
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::api::{
+ BrokerAdvertise, ERROR_INVALID_CONFIG, ERROR_INVALID_REQUEST, ERROR_INVALID_TOPIC_EXCEPTION,
+ ERROR_NONE, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState,
+};
+use iggy_gateway_kafka::protocol::handlers::{alter_configs, describe_configs};
+
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+
+use iggy_server::TestServer;
+
+const DESCRIBE_V4: i16 = 4;
+const DESCRIBE_V1: i16 = 1;
+const ALTER_V2: i16 = 2;
+const ALTER_V0: i16 = 0;
+const TOPIC: i8 = 2;
+const SOURCE_TOPIC: i8 = 1;
+const SOURCE_DEFAULT: i8 = 5;
+
+const fn text(value: &'static str) -> StrBytes {
+ StrBytes::from_static_str(value)
+}
+
+fn encode<M: Encodable>(message: &M, version: i16) -> Bytes {
+ let mut buf = BytesMut::new();
+ message.encode(&mut buf, version).expect("encode request");
+ buf.freeze()
+}
+
+fn decode<M: Decodable>(body: Bytes, version: i16) -> M {
+ let mut buf = body;
+ M::decode(&mut buf, version).expect("decode response")
+}
+
+async fn connected_state(server: &TestServer) -> GatewayState {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge connects");
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ 8 * 1024 * 1024,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+}
+
+async fn create_topic(state: &GatewayState, name: &str) {
+ let bridge = state.bridge.as_ref().expect("bridge");
+ bridge
+ .ensure_stream_and_topic(name, 1)
+ .await
+ .expect("create topic");
+}
+
+fn describe_resource(name: &str, keys: Option<Vec<&'static str>>) -> DescribeConfigsResource {
+ DescribeConfigsResource::default()
+ .with_resource_type(TOPIC)
+ .with_resource_name(StrBytes::from(name.to_string()))
+ .with_configuration_keys(keys.map(|keys| keys.into_iter().map(text).collect()))
+}
+
+async fn describe(
+ state: &GatewayState,
+ version: i16,
+ resources: Vec<DescribeConfigsResource>,
+ include_synonyms: bool,
+ include_documentation: bool,
+) -> DescribeConfigsResponse {
+ let request = DescribeConfigsRequest::default()
+ .with_resources(resources)
+ .with_include_synonyms(include_synonyms)
+ .with_include_documentation(include_documentation);
+ let outcome = describe_configs::handle(state, version, encode(&request, version)).await;
+ decode(outcome.expect_response("DescribeConfigs answers"), version)
+}
+
+fn alter_resource(
+ name: &str,
+ resource_type: i8,
+ configs: &[(&'static str, &'static str)],
+) -> AlterConfigsResource {
+ AlterConfigsResource::default()
+ .with_resource_type(resource_type)
+ .with_resource_name(StrBytes::from(name.to_string()))
+ .with_configs(
+ configs
+ .iter()
+ .map(|(key, value)| {
+ AlterableConfig::default()
+ .with_name(text(key))
+ .with_value(Some(text(value)))
+ })
+ .collect(),
+ )
+}
+
+async fn alter(
+ state: &GatewayState,
+ version: i16,
+ resources: Vec<AlterConfigsResource>,
+ validate_only: bool,
+) -> AlterConfigsResponse {
+ let request = AlterConfigsRequest::default()
+ .with_resources(resources)
+ .with_validate_only(validate_only);
+ let outcome = alter_configs::handle(state, version, encode(&request, version)).await;
+ decode(outcome.expect_response("AlterConfigs answers"), version)
+}
+
+fn entry<'a>(
+ configs: &'a [DescribeConfigsResourceResult],
+ name: &str,
+) -> &'a DescribeConfigsResourceResult {
+ configs
+ .iter()
+ .find(|config| config.name.as_str() == name)
+ .unwrap_or_else(|| panic!("missing config {name}"))
+}
+
+#[tokio::test]
+#[serial]
+async fn describe_configs_returns_retention_and_cleanup_and_alter_persists_retention() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ create_topic(&state, "orders").await;
+
+ let described = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", None)],
+ true,
+ true,
+ )
+ .await;
+ let result = &described.results[0];
+ assert_eq!(result.error_code, ERROR_NONE);
+ assert_eq!(result.configs.len(), 2);
+ let retention = entry(&result.configs, "retention.ms");
+ assert_eq!(retention.value.as_ref().map(StrBytes::as_str), Some("-1"));
+ assert!(!retention.read_only);
+ assert!(!retention.is_sensitive);
+ assert_eq!(retention.config_source, SOURCE_DEFAULT);
+ assert!(retention.synonyms.is_empty());
+ assert!(retention.documentation.is_some());
+ let cleanup = entry(&result.configs, "cleanup.policy");
+ assert_eq!(cleanup.value.as_ref().map(StrBytes::as_str), Some("delete"));
+ assert!(cleanup.read_only);
+ assert_eq!(cleanup.config_source, SOURCE_DEFAULT);
+ assert!(cleanup.synonyms.is_empty());
+ assert!(cleanup.documentation.is_some());
+
+ let without_docs = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", None)],
+ false,
+ false,
+ )
+ .await;
+ let bare = &without_docs.results[0].configs;
+ assert!(entry(bare, "retention.ms").documentation.is_none());
+ assert!(entry(bare, "retention.ms").synonyms.is_empty());
+
+ let filtered = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", Some(vec!["cleanup.policy"]))],
+ false,
+ true,
+ )
+ .await;
+ assert_eq!(filtered.results[0].error_code, ERROR_NONE);
+ assert_eq!(filtered.results[0].configs.len(), 1);
+ assert_eq!(
+ filtered.results[0].configs[0].name.as_str(),
+ "cleanup.policy"
+ );
+
+ let v1 = describe(
+ &state,
+ DESCRIBE_V1,
+ vec![describe_resource("orders", None)],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(v1.results[0].error_code, ERROR_NONE);
+ assert_eq!(v1.results[0].configs.len(), 2);
+
+ let validated = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource("orders", TOPIC, &[("retention.ms", "8000")])],
+ true,
+ )
+ .await;
+ assert_eq!(validated.responses[0].error_code, ERROR_NONE);
+ let still_default = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", Some(vec!["retention.ms"]))],
+ false,
+ false,
+ )
+ .await;
+ let unchanged = &still_default.results[0].configs[0];
+ assert_eq!(unchanged.value.as_ref().map(StrBytes::as_str), Some("-1"));
+ assert_eq!(unchanged.config_source, SOURCE_DEFAULT);
+
+ let written = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource("orders", TOPIC, &[("retention.ms", "8000")])],
+ false,
+ )
+ .await;
+ assert_eq!(written.responses[0].error_code, ERROR_NONE);
+ let after = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", Some(vec!["retention.ms"]))],
+ false,
+ false,
+ )
+ .await;
+ let changed = &after.results[0].configs[0];
+ assert_eq!(changed.value.as_ref().map(StrBytes::as_str), Some("8000"));
+ assert_eq!(changed.config_source, SOURCE_TOPIC);
+
+ let cleared = alter(
+ &state,
+ ALTER_V0,
+ vec![alter_resource("orders", TOPIC, &[("retention.ms", "-1")])],
+ false,
+ )
+ .await;
+ assert_eq!(cleared.responses[0].error_code, ERROR_NONE);
+ let never = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", Some(vec!["retention.ms"]))],
+ false,
+ false,
+ )
+ .await;
+ let cleared_entry = &never.results[0].configs[0];
+ assert_eq!(
+ cleared_entry.value.as_ref().map(StrBytes::as_str),
+ Some("-1")
+ );
+ assert_eq!(cleared_entry.config_source, SOURCE_TOPIC);
+}
+
+#[tokio::test]
+#[serial]
+async fn describe_and_alter_configs_reject_unknown_nontopic_missing_and_invalid_names() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ create_topic(&state, "orders").await;
+
+ let unknown = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource(
+ "orders",
+ Some(vec!["retention.ms", "no.such"]),
+ )],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(unknown.results[0].error_code, ERROR_INVALID_CONFIG);
+ assert_eq!(unknown.results[0].configs.len(), 2);
+ assert_eq!(unknown.results[0].configs[1].name.as_str(), "no.such");
+ let message = unknown.results[0]
+ .error_message
+ .as_ref()
+ .map_or("", StrBytes::as_str);
+ assert!(!message.contains("no.such"));
+
+ let broker = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![
+ DescribeConfigsResource::default()
+ .with_resource_type(4)
+ .with_resource_name(StrBytes::from_static_str("1"))
+ .with_configuration_keys(None),
+ describe_resource("orders", Some(vec!["cleanup.policy"])),
+ ],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(broker.results[0].error_code, ERROR_INVALID_REQUEST);
+ assert!(broker.results[0].configs.is_empty());
+ assert_eq!(
+ broker.results[0]
+ .error_message
+ .as_ref()
+ .map(StrBytes::as_str),
+ Some("only topic resources are supported")
+ );
+ assert_eq!(broker.results[1].error_code, ERROR_NONE);
+
+ let missing = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("missing-topic", None)],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(
+ missing.results[0].error_code,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION
+ );
+ assert!(missing.results[0].configs.is_empty());
+
+ let invalid = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("has a space", None)],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(invalid.results[0].error_code, ERROR_INVALID_TOPIC_EXCEPTION);
+ let reason = invalid.results[0]
+ .error_message
+ .as_ref()
+ .map_or("", StrBytes::as_str);
+ assert!(!reason.contains("has a space"));
+
+ let rejected = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource("orders", TOPIC, &[("no.such", "1")])],
+ false,
+ )
+ .await;
+ assert_eq!(rejected.responses[0].error_code, ERROR_INVALID_CONFIG);
+ let still = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", Some(vec!["retention.ms"]))],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(
+ still.results[0].configs[0]
+ .value
+ .as_ref()
+ .map(StrBytes::as_str),
+ Some("-1")
+ );
+ assert_eq!(still.results[0].configs[0].config_source, SOURCE_DEFAULT);
+
+ let readonly = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource(
+ "orders",
+ TOPIC,
+ &[("cleanup.policy", "compact")],
+ )],
+ false,
+ )
+ .await;
+ assert_eq!(readonly.responses[0].error_code, ERROR_INVALID_CONFIG);
+
+ let missing_alter = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource(
+ "missing-topic",
+ TOPIC,
+ &[("retention.ms", "1000")],
+ )],
+ false,
+ )
+ .await;
+ assert_eq!(
+ missing_alter.responses[0].error_code,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION
+ );
+
+ let bad_name = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource(
+ "has a space",
+ TOPIC,
+ &[("retention.ms", "1000")],
+ )],
+ false,
+ )
+ .await;
+ assert_eq!(
+ bad_name.responses[0].error_code,
+ ERROR_INVALID_TOPIC_EXCEPTION
+ );
+ let alter_reason = bad_name.responses[0]
+ .error_message
+ .as_ref()
+ .map_or("", StrBytes::as_str);
+ assert!(!alter_reason.contains("has a space"));
+
+ let nontopic = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource("1", 4, &[("retention.ms", "1000")])],
+ false,
+ )
+ .await;
+ assert_eq!(nontopic.responses[0].error_code, ERROR_INVALID_REQUEST);
+ assert_eq!(
+ nontopic.responses[0]
+ .error_message
+ .as_ref()
+ .map(StrBytes::as_str),
+ Some("only topic resources are supported")
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn alter_configs_one_bad_key_skips_only_that_resource() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ create_topic(&state, "kept").await;
+ create_topic(&state, "changed").await;
+
+ let altered = alter(
+ &state,
+ ALTER_V2,
+ vec![
+ alter_resource("kept", TOPIC, &[("retention.ms", "4000"), ("no.such", "1")]),
+ alter_resource("changed", TOPIC, &[("retention.ms", "5000")]),
+ ],
+ false,
+ )
+ .await;
+ assert_eq!(altered.responses[0].error_code, ERROR_INVALID_CONFIG);
+ assert_eq!(altered.responses[1].error_code, ERROR_NONE);
+
+ let described = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![
+ describe_resource("kept", Some(vec!["retention.ms"])),
+ describe_resource("changed", Some(vec!["retention.ms"])),
+ ],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(
+ described.results[0].configs[0]
+ .value
+ .as_ref()
+ .map(StrBytes::as_str),
+ Some("-1")
+ );
+ assert_eq!(
+ described.results[0].configs[0].config_source,
+ SOURCE_DEFAULT
+ );
+ assert_eq!(
+ described.results[1].configs[0]
+ .value
+ .as_ref()
+ .map(StrBytes::as_str),
+ Some("5000")
+ );
+ assert_eq!(described.results[1].configs[0].config_source, SOURCE_TOPIC);
+}
diff --git a/gateways/kafka/tests/java/ConfigsProbe.java b/gateways/kafka/tests/java/ConfigsProbe.java
new file mode 100644
index 000000000..6c757f404
--- /dev/null
+++ b/gateways/kafka/tests/java/ConfigsProbe.java
@@ -0,0 +1,129 @@
+/*
+ * 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.
+ */
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.Config;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
+
+/**
+ * Kafka 3.9 AdminClient probe for DescribeConfigs and the deprecated AlterConfigs API.
+ * Prints elapsed time for 50 describes and does not fail on a latency threshold.
+ */
+public final class ConfigsProbe {
+ private ConfigsProbe() {}
+
+ @SuppressWarnings("deprecation")
+ public static void main(String[] args) throws Exception {
+ if (args.length != 2) {
+ throw new IllegalArgumentException("usage: ConfigsProbe <bootstrap> <topic>");
+ }
+ String bootstrap = args[0];
+ String topic = args[1];
+ Properties props = new Properties();
+ props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+ props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "20000");
+ props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "20000");
+ props.put(AdminClientConfig.CLIENT_ID_CONFIG, "iggy-configs-probe");
+
+ ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
+ try (AdminClient admin = AdminClient.create(props)) {
+ Config before = describe(admin, resource);
+ ConfigEntry retention = before.get("retention.ms");
+ ConfigEntry cleanup = before.get("cleanup.policy");
+ require(retention != null && "-1".equals(retention.value()), "initial retention.ms");
+ require(
+ retention.source() == ConfigEntry.ConfigSource.DEFAULT_CONFIG,
+ "initial retention source");
+ require(!retention.isReadOnly(), "retention.ms is writable");
+ require(cleanup != null && "delete".equals(cleanup.value()), "cleanup.policy");
+ require(cleanup.isReadOnly(), "cleanup.policy is read only");
+ require(
+ cleanup.source() == ConfigEntry.ConfigSource.DEFAULT_CONFIG,
+ "cleanup source");
+ System.out.println(
+ "retention.ms=" + retention.value() + " source=" + retention.source());
+ System.out.println(
+ "cleanup.policy="
+ + cleanup.value()
+ + " read_only="
+ + cleanup.isReadOnly()
+ + " source="
+ + cleanup.source());
+
+ try {
+ admin.alterConfigs(Map.of(resource, config("no.such", "1"))).all().get();
+ throw new AssertionError("unknown key was accepted");
+ } catch (ExecutionException exception) {
+ if (!(exception.getCause() instanceof InvalidConfigurationException)) {
+ throw exception;
+ }
+ System.out.println("unknown_key_rejected=" + exception.getCause().getClass().getSimpleName());
+ }
+
+ Config stillDefault = describe(admin, resource);
+ require("-1".equals(stillDefault.get("retention.ms").value()), "unknown key wrote retention");
+
+ admin.alterConfigs(Map.of(resource, config("retention.ms", "8000")), new AlterConfigsOptions().validateOnly(true))
+ .all()
+ .get();
+ Config validated = describe(admin, resource);
+ require("-1".equals(validated.get("retention.ms").value()), "validate_only wrote retention");
+ System.out.println("validate_only_retention.ms=" + validated.get("retention.ms").value());
+
+ admin.alterConfigs(Map.of(resource, config("retention.ms", "8000"))).all().get();
+ Config altered = describe(admin, resource);
+ ConfigEntry changed = altered.get("retention.ms");
+ require("8000".equals(changed.value()), "altered retention.ms");
+ require(
+ changed.source() == ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG,
+ "altered retention source");
+ System.out.println("altered_retention.ms=" + changed.value() + " source=" + changed.source());
+
+ long started = System.nanoTime();
+ for (int i = 0; i < 50; i++) {
+ describe(admin, resource);
+ }
+ long elapsedMs = (System.nanoTime() - started) / 1_000_000L;
+ System.out.println("describe_batch_50_elapsed_ms=" + elapsedMs);
+ System.out.println("CONFIGS_PROBE_OK");
+ }
+ }
+
+ private static Config describe(AdminClient admin, ConfigResource resource) throws Exception {
+ return admin.describeConfigs(List.of(resource)).all().get().get(resource);
+ }
+
+ private static Config config(String name, String value) {
+ return new Config(List.of(new ConfigEntry(name, value)));
+ }
+
+ private static void require(boolean condition, String message) {
+ if (!condition) {
+ throw new AssertionError(message);
+ }
+ }
+}