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
The following commit(s) were added to
refs/heads/feat/gateways-3546-describe-alter-configs by this push:
new bb7446772 Implementing review commands.
bb7446772 is described below
commit bb74467721e721a3b1ef6aa41c63be053ad47307
Author: ryerraguntla <[email protected]>
AuthorDate: Sun Oct 4 06:48:44 2026 -0400
Implementing review commands.
gateways/kafka/docs/CONFIGS.md
gateways/kafka/src/protocol/handlers/alter_configs.rs
gateways/kafka/src/protocol/handlers/describe_configs.rs
gateways/kafka/src/protocol/handlers/topic_config.rs
gateways/kafka/tests/common/wire.rs
gateways/kafka/tests/configs_handler_tests.rs
gateways/kafka/tests/configs_java_client_tests.rs
gateways/kafka/tests/configs_real_bridge_tests.rs
---
gateways/kafka/docs/CONFIGS.md | 9 +-
.../kafka/src/protocol/handlers/alter_configs.rs | 154 +++++----------
.../src/protocol/handlers/describe_configs.rs | 215 ++++++++++-----------
.../kafka/src/protocol/handlers/topic_config.rs | 96 +++++++--
gateways/kafka/tests/common/wire.rs | 16 +-
gateways/kafka/tests/configs_handler_tests.rs | 21 +-
gateways/kafka/tests/configs_java_client_tests.rs | 37 +++-
gateways/kafka/tests/configs_real_bridge_tests.rs | 141 ++++++++++++--
8 files changed, 424 insertions(+), 265 deletions(-)
diff --git a/gateways/kafka/docs/CONFIGS.md b/gateways/kafka/docs/CONFIGS.md
index b734ca0a1..4acf0a942 100644
--- a/gateways/kafka/docs/CONFIGS.md
+++ b/gateways/kafka/docs/CONFIGS.md
@@ -34,7 +34,9 @@ the never-expire default. The gateway does not round.
## What an alter stores
-Only `retention.ms` is written, through `update_topic`'s `message_expiry`.
+Only `retention.ms` is written, through `update_topic`'s `message_expiry`. A
resource
+whose `configs` list does not name `retention.ms` is left unchanged. Omitting
the key
+does not clear a previously set expiry.
- `-1` stores `IggyExpiry::NeverExpire`. The server treats `0`
(`IggyExpiry::ServerDefault`) as "leave the current expiry", so `-1` is not
@@ -63,5 +65,6 @@ call `update_topic`.
| 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 |
+| More than 100 distinct topic names | `POLICY_VIOLATION` (44) | Every
resource in the request |
+| A topic name repeated across resources | `INVALID_REQUEST` (42) | Every
occurrence of the duplicate, not just the second one. Same choice CreateTopics
makes for a repeated topic name |
+| Non-empty unknown tagged fields | `INVALID_REQUEST` (42) | Request-level
tags fail every resource. A resource or config entry's own tags fail that
resource. Empty tagged fields are the normal flexible encoding |
diff --git a/gateways/kafka/src/protocol/handlers/alter_configs.rs
b/gateways/kafka/src/protocol/handlers/alter_configs.rs
index 26154fb75..8da1671b3 100644
--- a/gateways/kafka/src/protocol/handlers/alter_configs.rs
+++ b/gateways/kafka/src/protocol/handlers/alter_configs.rs
@@ -17,8 +17,6 @@
//! `AlterConfigs` (API key 33).
-use std::collections::HashMap;
-
use bytes::Bytes;
use iggy::prelude::IggyError;
use kafka_protocol::messages::AlterConfigsRequest;
@@ -39,8 +37,8 @@ use crate::protocol::api::{
};
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,
+ 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,
@@ -73,17 +71,6 @@ pub async fn handle(state: &GatewayState, api_version: i16,
body: Bytes) -> Hand
}
};
- 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
@@ -129,12 +116,33 @@ async fn alter_all(
req: &AlterConfigsRequest,
deadline: Instant,
) -> Vec<AlterConfigsResourceResponse> {
- let topic_names = req
+ 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());
- if exceeds_topic_cap(topic_names) {
+ .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
@@ -144,14 +152,17 @@ async fn alter_all(
.collect();
}
- let mut seen: HashMap<String, AlterConfigsResourceResponse> =
HashMap::new();
+ // 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
- && let Some(prior) = seen.get(resource.resource_name.as_str())
+ && duplicate_names.contains(resource.resource_name.as_str())
{
- responses.push(prior.clone());
+ responses.push(resource_error(resource, ERROR_INVALID_REQUEST,
None));
continue;
}
if deadline_exceeded || Instant::now() >= deadline {
@@ -162,9 +173,7 @@ async fn alter_all(
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);
+ responses.push(resource_error(resource, ERROR_REQUEST_TIMED_OUT,
None));
continue;
}
@@ -172,22 +181,11 @@ async fn alter_all(
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,
@@ -203,7 +201,7 @@ async fn alter_one(
return resource_error(
resource,
ERROR_INVALID_REQUEST,
- Some(static_text(UNKNOWN_TAGGED_FIELDS)),
+ Some(static_text("unknown tagged fields are not supported")),
);
}
if resource.resource_type != RESOURCE_TYPE_TOPIC {
@@ -232,15 +230,23 @@ async fn alter_one(
);
return resource_error(resource, ERROR_REQUEST_TIMED_OUT, None);
}
- match bridge.get_kafka_topic(name).await {
- Ok(Some(_topic)) => {}
- Ok(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);
}
- Err(error) => {
- let (code, message) = bridge_failure(&error);
+ 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 expiry = match
plan_retention_update(resource.configs.iter().map(|config| {
@@ -258,12 +264,14 @@ async fn alter_one(
);
}
};
-
- // `validate_only` already performed the existence read and the same key
checks.
- // It does not call `update_topic`.
+ // A resource that does not name retention.ms has nothing to store.
Omitting the key
+ // does not clear a previously set expiry.
let Some(expiry) = expiry else {
return resource_error(resource, ERROR_NONE, None);
};
+
+ // `validate_only` already performed the existence read and the same key
checks.
+ // It does not call `update_topic`.
if validate_only {
return resource_error(resource, ERROR_NONE, None);
}
@@ -284,36 +292,12 @@ async fn alter_one(
resource_error(resource, ERROR_NONE, None)
}
Err(error) => {
- let (code, message) = bridge_failure(&error);
+ let (code, message) = bridge_failure(&error, "AlterConfigs",
"altering");
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,
@@ -325,35 +309,3 @@ fn resource_error(
.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
index 530b305a4..200261443 100644
--- a/gateways/kafka/src/protocol/handlers/describe_configs.rs
+++ b/gateways/kafka/src/protocol/handlers/describe_configs.rs
@@ -17,8 +17,6 @@
//! `DescribeConfigs` (API key 32).
-use std::collections::HashMap;
-
use bytes::Bytes;
use iggy::prelude::TopicDetails;
use kafka_protocol::messages::DescribeConfigsRequest;
@@ -30,8 +28,8 @@ use kafka_protocol::messages::describe_configs_response::{
use kafka_protocol::protocol::StrBytes;
use tokio::time::Instant;
+use crate::bridge::IggyBridge;
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,
@@ -41,9 +39,10 @@ use crate::protocol::api::{
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,
+ CONFIG_TYPE_LONG, CONFIG_TYPE_STRING, CONFIG_TYPE_UNKNOWN, ListedKey,
ONLY_TOPIC_RESOURCES,
+ RESOURCE_TYPE_TOPIC, RETENTION_DOC, RETENTION_MS, bridge_failure,
exceeds_topic_cap,
+ find_duplicate_names, listed_keys, message_expiry_is_explicit,
name_reason, retention_ms,
+ static_text, topic_cap_message,
};
use crate::protocol::handlers::{
decode_guarded, encode_message, is_supported_version, respond_or_close,
@@ -76,17 +75,6 @@ pub async fn handle(state: &GatewayState, api_version: i16,
body: Bytes) -> Hand
}
};
- 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
@@ -133,12 +121,39 @@ async fn describe_all(
req: &DescribeConfigsRequest,
deadline: Instant,
) -> Vec<DescribeConfigsResult> {
- let topic_names = req
+ 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());
- if exceeds_topic_cap(topic_names) {
+ .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
@@ -155,16 +170,23 @@ async fn describe_all(
.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();
+ // 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
- && let Some(prior) = seen.get(resource.resource_name.as_str())
+ && duplicate_names.contains(resource.resource_name.as_str())
{
- results.push(prior.clone());
+ results.push(resource_error(
+ resource,
+ ERROR_INVALID_REQUEST,
+ None,
+ Vec::new(),
+ ));
continue;
}
if deadline_exceeded || Instant::now() >= deadline {
@@ -175,9 +197,12 @@ async fn describe_all(
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);
+ results.push(resource_error(
+ resource,
+ ERROR_REQUEST_TIMED_OUT,
+ None,
+ Vec::new(),
+ ));
continue;
}
@@ -192,22 +217,11 @@ async fn describe_all(
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,
@@ -219,7 +233,7 @@ async fn describe_one(
return resource_error(
resource,
ERROR_INVALID_REQUEST,
- Some(static_text(UNKNOWN_TAGGED_FIELDS)),
+ Some(static_text("unknown tagged fields are not supported")),
Vec::new(),
);
}
@@ -248,34 +262,49 @@ async fn describe_one(
);
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) => {
+ 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());
}
- Err(error) => {
- let (code, message) = bridge_failure(&error, "reading");
+ 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 (configs, unknown, unrepresentable) = described_entries(
+ let (configs, unknown_key, 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"
- };
+ if let Some(unknown_key) = unknown_key {
return resource_error(
resource,
ERROR_INVALID_CONFIG,
- Some(static_text(message)),
+ Some(StrBytes::from(format!(
+ "unknown config key '{unknown_key}'"
+ ))),
+ configs,
+ );
+ }
+ if unrepresentable {
+ return resource_error(
+ resource,
+ ERROR_INVALID_CONFIG,
+ Some(static_text(
+ "stored message expiry is not a whole number of milliseconds",
+ )),
configs,
);
}
@@ -288,11 +317,11 @@ fn described_entries(
configuration_keys: Option<&[StrBytes]>,
include_synonyms: bool,
include_documentation: bool,
-) -> (Vec<DescribeConfigsResourceResult>, bool, bool) {
+) -> (Vec<DescribeConfigsResourceResult>, Option<String>, 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 unknown_key = None;
let mut unrepresentable = false;
let mut configs = Vec::with_capacity(keys.len());
for key in keys {
@@ -327,20 +356,22 @@ fn described_entries(
include_documentation.then_some(CLEANUP_DOC),
)),
ListedKey::Unknown(name) => {
- unknown = true;
+ if unknown_key.is_none() {
+ unknown_key = Some(name.clone());
+ }
configs.push(config_entry(
&name,
None,
false,
CONFIG_SOURCE_DEFAULT,
- 0,
+ CONFIG_TYPE_UNKNOWN,
include_synonyms,
None,
));
}
}
}
- (configs, unknown, unrepresentable)
+ (configs, unknown_key, unrepresentable)
}
fn config_entry(
@@ -363,33 +394,17 @@ fn config_entry(
.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()),
- ];
+/// Neither known key has a synonym. `true` asks for the list and receives
none.
+#[allow(
+ clippy::if_same_then_else,
+ reason = "true asks for synonyms and there are none"
+)]
+fn synonyms(include_synonyms: bool) -> Vec<DescribeConfigsSynonym> {
+ if include_synonyms {
+ Vec::new()
+ } else {
+ Vec::new()
}
- resources
- .iter()
- .map(|resource| {
- resource_error(
- resource,
- ERROR_INVALID_REQUEST,
- Some(static_text(UNKNOWN_TAGGED_FIELDS)),
- Vec::new(),
- )
- })
- .collect()
}
fn resource_error(
@@ -405,35 +420,3 @@ fn resource_error(
.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
index 876092282..1b6ccb56c 100644
--- a/gateways/kafka/src/protocol/handlers/topic_config.rs
+++ b/gateways/kafka/src/protocol/handlers/topic_config.rs
@@ -30,6 +30,8 @@ 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;
@@ -45,12 +47,14 @@ 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 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.";
@@ -58,8 +62,10 @@ pub const CLEANUP_DOC: &str = "Iggy deletes expired messages
and does not compac
/// 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.
+/// 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.
@@ -139,12 +145,8 @@ pub fn retention_ms(expiry: IggyExpiry, explicit: bool) ->
Result<RetentionMs, (
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(),
+ value: (micros / 1_000).to_string(),
source: CONFIG_SOURCE_TOPIC,
})
}
@@ -177,9 +179,9 @@ pub fn parse_retention_ms(value: &str) ->
Result<IggyExpiry, ConfigFault> {
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.
+ // `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);
}
@@ -203,6 +205,67 @@ pub fn topic_cap_message(api_name: &str) -> String {
)
}
+/// 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 both known keys. Any other name is
@@ -327,6 +390,17 @@ mod tests {
);
}
+ #[test]
+ fn find_duplicate_names_finds_a_name_repeated_across_two_resources() {
+ let duplicates = find_duplicate_names(["orders", "payments",
"orders"]);
+ assert_eq!(duplicates, HashSet::from(["orders"]));
+ }
+
+ #[test]
+ fn find_duplicate_names_is_empty_when_every_name_is_unique() {
+ assert!(find_duplicate_names(["orders", "payments"]).is_empty());
+ }
+
#[test]
fn one_bad_key_rejects_the_whole_alter_plan() {
let err = plan_retention_update([(RETENTION_MS, Some("5000")),
("no.such", Some("1"))])
diff --git a/gateways/kafka/tests/common/wire.rs
b/gateways/kafka/tests/common/wire.rs
index 1605f1885..b9dd3c6bb 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -26,10 +26,24 @@
//! every function some *other* binary uses as unused here.
#![allow(dead_code)]
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
+use kafka_protocol::protocol::{Decodable, Encodable};
use super::codec::Encoder;
+/// Encodes any `kafka_protocol` request or response type at `version`.
+pub fn encode<M: Encodable>(message: &M, version: i16) -> Bytes {
+ let mut buf = BytesMut::new();
+ message.encode(&mut buf, version).expect("encode request");
+ buf.freeze()
+}
+
+/// Decodes any `kafka_protocol` request or response type at `version`.
+pub fn decode<M: Decodable>(body: Bytes, version: i16) -> M {
+ let mut buf = body;
+ M::decode(&mut buf, version).expect("decode response")
+}
+
/// Consumer-group and admin keys explicitly out of scope in SCOPE.md.
pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] = &[
(8, "OffsetCommit"),
diff --git a/gateways/kafka/tests/configs_handler_tests.rs
b/gateways/kafka/tests/configs_handler_tests.rs
index ede44539b..4bf8c9346 100644
--- a/gateways/kafka/tests/configs_handler_tests.rs
+++ b/gateways/kafka/tests/configs_handler_tests.rs
@@ -17,7 +17,6 @@
//! `DescribeConfigs` and `AlterConfigs` with the bridge off.
-use bytes::{Bytes, BytesMut};
use kafka_protocol::messages::alter_configs_request::{
AlterConfigsRequest, AlterConfigsResource, AlterableConfig,
};
@@ -26,13 +25,20 @@ 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 kafka_protocol::protocol::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};
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/wire.rs"]
+mod wire;
+
+use wire::{decode, encode};
+
const DESCRIBE_VERSION: i16 = 4;
const ALTER_VERSION: i16 = 2;
@@ -47,17 +53,6 @@ fn state_without_bridge() -> GatewayState {
)
}
-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()
diff --git a/gateways/kafka/tests/configs_java_client_tests.rs
b/gateways/kafka/tests/configs_java_client_tests.rs
index 4dc590508..1fde0a4ec 100644
--- a/gateways/kafka/tests/configs_java_client_tests.rs
+++ b/gateways/kafka/tests/configs_java_client_tests.rs
@@ -46,6 +46,34 @@ const TOPIC: &str = "orders";
/// 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())
@@ -188,12 +216,9 @@ async fn spawn_gateway(server: &TestServer) -> SocketAddr {
#[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)
- );
+ if java_missing() {
+ return;
+ }
let jars = jar_dir();
ensure_jars(&jars);
diff --git a/gateways/kafka/tests/configs_real_bridge_tests.rs
b/gateways/kafka/tests/configs_real_bridge_tests.rs
index 933a85a9f..8893c6ee7 100644
--- a/gateways/kafka/tests/configs_real_bridge_tests.rs
+++ b/gateways/kafka/tests/configs_real_bridge_tests.rs
@@ -19,7 +19,6 @@
use std::sync::Arc;
-use bytes::{Bytes, BytesMut};
use kafka_protocol::messages::alter_configs_request::{
AlterConfigsRequest, AlterConfigsResource, AlterableConfig,
};
@@ -30,7 +29,7 @@ use kafka_protocol::messages::describe_configs_request::{
use kafka_protocol::messages::describe_configs_response::{
DescribeConfigsResourceResult, DescribeConfigsResponse,
};
-use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
+use kafka_protocol::protocol::StrBytes;
use serial_test::serial;
use tokio_util::sync::CancellationToken;
@@ -42,10 +41,15 @@ use iggy_gateway_kafka::protocol::api::{
};
use iggy_gateway_kafka::protocol::handlers::{alter_configs, describe_configs};
+#[path = "common/codec.rs"]
+mod codec;
#[path = "common/iggy_server.rs"]
mod iggy_server;
+#[path = "common/wire.rs"]
+mod wire;
use iggy_server::TestServer;
+use wire::{decode, encode};
const DESCRIBE_V4: i16 = 4;
const DESCRIBE_V1: i16 = 1;
@@ -59,17 +63,6 @@ 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
@@ -318,7 +311,10 @@ async fn
describe_and_alter_configs_reject_unknown_nontopic_missing_and_invalid_
.error_message
.as_ref()
.map_or("", StrBytes::as_str);
- assert!(!message.contains("no.such"));
+ assert!(
+ message.contains("no.such"),
+ "unknown-key message must name the key, matching AlterConfigs' own
wording: {message}"
+ );
let broker = describe(
&state,
@@ -519,3 +515,120 @@ async fn
alter_configs_one_bad_key_skips_only_that_resource() {
);
assert_eq!(described.results[1].configs[0].config_source, SOURCE_TOPIC);
}
+
+/// Regression test for the duplicate-resource memoization bug: a second
`AlterConfigs`
+/// resource entry for an already-seen topic name must not be silently
answered from a cached
+/// result - its own, different `retention.ms` must never be reported as
applied when the write
+/// for it never ran.
+#[tokio::test]
+#[serial]
+async fn alter_configs_rejects_every_occurrence_of_a_duplicate_resource_name()
{
+ 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 altered = alter(
+ &state,
+ ALTER_V2,
+ vec![
+ alter_resource("orders", TOPIC, &[("retention.ms", "4000")]),
+ alter_resource("orders", TOPIC, &[("retention.ms", "5000")]),
+ ],
+ false,
+ )
+ .await;
+ assert_eq!(altered.responses.len(), 2);
+ assert_eq!(altered.responses[0].error_code, ERROR_INVALID_REQUEST);
+ assert_eq!(altered.responses[1].error_code, ERROR_INVALID_REQUEST);
+
+ // Neither occurrence's value was applied - the stored retention is still
the default.
+ let described = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", Some(vec!["retention.ms"]))],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(
+ described.results[0].configs[0]
+ .value
+ .as_ref()
+ .map(StrBytes::as_str),
+ Some("-1")
+ );
+}
+
+/// The `DescribeConfigs` sibling of the same memoization pattern: a duplicate
resource name
+/// with a *different* `configuration_keys` filter must not be answered from
the first entry's
+/// cached, differently-filtered result.
+#[tokio::test]
+#[serial]
+async fn
describe_configs_rejects_every_occurrence_of_a_duplicate_resource_name() {
+ 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", Some(vec!["retention.ms"])),
+ describe_resource("orders", Some(vec!["cleanup.policy"])),
+ ],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(described.results.len(), 2);
+ assert_eq!(described.results[0].error_code, ERROR_INVALID_REQUEST);
+ assert_eq!(described.results[1].error_code, ERROR_INVALID_REQUEST);
+}
+
+/// An alter that does not name `retention.ms` does not clear a previously set
expiry.
+#[tokio::test]
+#[serial]
+async fn alter_configs_with_no_configs_leaves_retention_unchanged() {
+ 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 first = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource("orders", TOPIC, &[("retention.ms", "4000")])],
+ false,
+ )
+ .await;
+ assert_eq!(first.responses[0].error_code, ERROR_NONE);
+
+ let reset = alter(
+ &state,
+ ALTER_V2,
+ vec![alter_resource("orders", TOPIC, &[])],
+ false,
+ )
+ .await;
+ assert_eq!(reset.responses[0].error_code, ERROR_NONE);
+
+ let described = describe(
+ &state,
+ DESCRIBE_V4,
+ vec![describe_resource("orders", Some(vec!["retention.ms"]))],
+ false,
+ false,
+ )
+ .await;
+ assert_eq!(
+ described.results[0].configs[0]
+ .value
+ .as_ref()
+ .map(StrBytes::as_str),
+ Some("4000"),
+ "an empty configs list must not clear a previously set retention.ms"
+ );
+ assert_eq!(described.results[0].configs[0].config_source, SOURCE_TOPIC);
+}
