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/03 19:05:04 UTC
(iggy) 01/01: Initial version of Delete Topics implementation
This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a commit to branch feat/gateways-3546-delete-describe-alter-configs
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8afc2c8e084a3a91369be0a2608fe133d94fdab5
Author: ryerraguntla <ra...@gmail.com>
AuthorDate: Sat Oct 3 14:52:12 2026 -0400
Initial version of Delete Topics implementation
---
.../kafka/src/protocol/handlers/delete_topics.rs | 272 +++++++++++++++++
.../kafka/tests/delete_topics_real_bridge_tests.rs | 322 +++++++++++++++++++++
2 files changed, 594 insertions(+)
diff --git a/gateways/kafka/src/protocol/handlers/delete_topics.rs b/gateways/kafka/src/protocol/handlers/delete_topics.rs
new file mode 100644
index 000000000..58f427614
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/delete_topics.rs
@@ -0,0 +1,272 @@
+// 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.
+
+//! `DeleteTopics` (API key 20).
+//!
+//! Deletes the Iggy topic a Kafka topic name resolves to. Never deletes the backing Iggy
+//! stream - see [`IggyBridge::delete_kafka_topic`]'s own doc comment for why: the shared
+//! `default_stream` can hold other Kafka topics' data, and this bridge has no way to tell
+//! whether the stream it just emptied is actually abandoned or just temporarily topic-less.
+
+use std::time::Duration;
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_topics_response::DeletableTopicResult;
+use kafka_protocol::messages::{DeleteTopicsRequest, DeleteTopicsResponse, TopicName};
+
+use tokio::time::Instant;
+
+use crate::bridge::{BridgeError, IggyBridge};
+use crate::error::Result;
+use crate::protocol::api::{
+ API_KEY_DELETE_TOPICS, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_NONE,
+ ERROR_NOT_CONTROLLER, ERROR_POLICY_VIOLATION, ERROR_REQUEST_TIMED_OUT, GatewayState,
+ HandleOutcome,
+};
+use crate::protocol::bounds_guard::validate_delete_topics_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, handle_versioned_request, is_supported_version,
+ respond_or_close, unsupported_version_response,
+};
+
+/// This bridge never advertises v6 (the `topics: Vec<DeleteTopicState>`, topic-id-based shape).
+///
+/// It has no concept of a Kafka topic id, only the Kafka-side name `TopicMapping` resolves. v1
+/// is the crate's own floor (`kafka_protocol` does not implement v0 for this message).
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_TOPICS,
+ min_version: 1,
+ max_version: 5,
+};
+
+/// Cap on distinct topic names one `DeleteTopics` request may address through the bridge.
+///
+/// Same rationale as `create_topics::MAX_BRIDGE_BACKED_TOPICS`: each name costs a bridge round
+/// trip against the single lockstep `IggyClient` every Kafka connection on this gateway shares.
+const MAX_BRIDGE_BACKED_TOPICS: usize = 100;
+
+/// Same bounds as `create_topics::clamp_request_timeout` - this request carries the identical
+/// KIP-4 `timeout_ms` field for the identical reason (a client-supplied deadline that becomes
+/// the aggregate bridge-work budget, otherwise unchecked).
+const MIN_REQUEST_TIMEOUT: Duration = Duration::from_millis(1_000);
+const MAX_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
+fn clamp_request_timeout(timeout_ms: i32) -> Duration {
+ let requested = Duration::from_millis(u64::try_from(timeout_ms).unwrap_or(0));
+ requested.clamp(MIN_REQUEST_TIMEOUT, MAX_REQUEST_TIMEOUT)
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ let Some(bridge) = &state.bridge else {
+ return handle_versioned_request(
+ API_KEY_DELETE_TOPICS,
+ api_version,
+ body,
+ |v, b| {
+ decode_guarded::<DeleteTopicsRequest>(v, b, |v, b| {
+ validate_delete_topics_shape(v, b, state.max_frame_size)
+ })
+ },
+ encode_response,
+ encode_error_response,
+ "DeleteTopics",
+ );
+ };
+
+ if !is_supported_version(API_KEY_DELETE_TOPICS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_TOPICS, api_version, |version| {
+ encode_error_response(version, ERROR_INVALID_REQUEST)
+ });
+ }
+
+ let req = match decode_guarded::<DeleteTopicsRequest>(api_version, body, |v, b| {
+ validate_delete_topics_shape(v, b, state.max_frame_size)
+ }) {
+ Ok(req) => req,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode DeleteTopics request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "DeleteTopics",
+ );
+ }
+ };
+
+ // RANGE caps at v5, so req.topics (the v6 topic-id shape) is always empty - topic_names is
+ // the only populated field at any version this bridge advertises.
+ if req.topic_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = req.topic_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "DeleteTopics request addresses too many topics; rejecting"
+ );
+ let message = kafka_protocol::protocol::StrBytes::from(format!(
+ "this gateway addresses at most {MAX_BRIDGE_BACKED_TOPICS} distinct topics per DeleteTopics request"
+ ));
+ let results = req
+ .topic_names
+ .iter()
+ .map(|name| {
+ DeletableTopicResult::default()
+ .with_name(Some(name.clone()))
+ .with_error_code(ERROR_POLICY_VIOLATION)
+ .with_error_message(Some(message.clone()))
+ })
+ .collect();
+ let resp = DeleteTopicsResponse::default().with_responses(results);
+ return respond_or_close(encode_message(&resp, api_version, 256), "DeleteTopics");
+ }
+
+ let deadline = Instant::now() + clamp_request_timeout(req.timeout_ms);
+ let results = delete_all_topics(bridge, &req.topic_names, deadline).await;
+ let resp = DeleteTopicsResponse::default().with_responses(results);
+ respond_or_close(encode_message(&resp, api_version, 256), "DeleteTopics")
+}
+
+/// Deletes every requested topic, independently of the others.
+///
+/// `deadline` bounds each topic's own bridge work individually (`timeout_at`), not the whole
+/// loop - same reasoning as `create_topics::create_all_topics`: a single `timeout` around the
+/// entire call would discard every already-resolved result the moment one topic's call ran
+/// long, answering a retriable code even for topics that had already deleted cleanly.
+///
+/// A name repeated within the same request is not specially rejected the way `CreateTopics`
+/// rejects a duplicate: deleting the same topic twice has no race to protect against the way
+/// creating it twice does - the first occurrence deletes it, the second then answers
+/// `UNKNOWN_TOPIC_OR_PARTITION` because by the time it runs, the topic genuinely doesn't exist
+/// any more. That's the correct answer for a duplicate, not a special case.
+async fn delete_all_topics(
+ bridge: &IggyBridge,
+ topic_names: &[TopicName],
+ deadline: Instant,
+) -> Vec<DeletableTopicResult> {
+ let mut results = Vec::with_capacity(topic_names.len());
+ for name in topic_names {
+ let result = match tokio::time::timeout_at(deadline, delete_one_topic(bridge, name)).await {
+ Ok(result) => result,
+ Err(_elapsed) => {
+ tracing::warn!(
+ kafka_topic = name.as_str(),
+ "DeleteTopics: this topic's bridge work exceeded the request deadline; \
+ answering retriable instead of blocking further"
+ );
+ DeletableTopicResult::default()
+ .with_name(Some(name.clone()))
+ .with_error_code(ERROR_REQUEST_TIMED_OUT)
+ .with_error_message(None)
+ }
+ };
+ results.push(result);
+ }
+ results
+}
+
+async fn delete_one_topic(bridge: &IggyBridge, name: &TopicName) -> DeletableTopicResult {
+ let result = DeletableTopicResult::default()
+ .with_name(Some(name.clone()))
+ .with_error_message(None);
+ match bridge.delete_kafka_topic(name.as_str()).await {
+ Ok(()) => result.with_error_code(ERROR_NONE),
+ Err(err) => bridge_error_result(result, name.as_str(), &err),
+ }
+}
+
+/// Maps a bridge failure to a Kafka result, logging the real cause server-side.
+///
+/// Same caution as `create_topics::bridge_error_result` on forwarding `err.to_string()`: a
+/// server-side rejection can reconstruct with default-valued fields, so only the client-caused
+/// variant (an invalid name) forwards its own text; everything else gets a fixed message.
+fn bridge_error_result(
+ result: DeletableTopicResult,
+ kafka_topic: &str,
+ err: &BridgeError,
+) -> DeletableTopicResult {
+ let error_code = err.to_kafka_error_code();
+ match err {
+ BridgeError::InvalidKafkaTopicName { reason, .. } => {
+ tracing::debug!(
+ kafka_topic,
+ reason,
+ "DeleteTopics rejected an invalid topic name"
+ );
+ result.with_error_code(error_code).with_error_message(Some(
+ kafka_protocol::protocol::StrBytes::from(reason.clone()),
+ ))
+ }
+ other => {
+ tracing::error!(kafka_topic, %other, "DeleteTopics failed against the Iggy bridge");
+ result.with_error_code(error_code).with_error_message(Some(
+ kafka_protocol::protocol::StrBytes::from(
+ "internal error deleting this topic".to_string(),
+ ),
+ ))
+ }
+ }
+}
+
+/// Well-formed `DeleteTopics` response naming every requested topic.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
+ encode_inner(version, &[], error_code)
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, req: &DeleteTopicsRequest) -> Result<Bytes> {
+ encode_inner(version, &req.topic_names, ERROR_NOT_CONTROLLER)
+}
+
+fn encode_inner(version: i16, topic_names: &[TopicName], forced_error: i16) -> Result<Bytes> {
+ let results = topic_names
+ .iter()
+ .map(|name| {
+ DeletableTopicResult::default()
+ .with_name(Some(name.clone()))
+ .with_error_code(forced_error)
+ .with_error_message(None)
+ })
+ .collect();
+ let resp = DeleteTopicsResponse::default().with_responses(results);
+ encode_message(&resp, version, 256)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn clamp_request_timeout_rejects_a_zero_or_negative_value_up_to_the_floor() {
+ assert_eq!(clamp_request_timeout(0), MIN_REQUEST_TIMEOUT);
+ assert_eq!(clamp_request_timeout(-1), MIN_REQUEST_TIMEOUT);
+ assert_eq!(clamp_request_timeout(i32::MIN), MIN_REQUEST_TIMEOUT);
+ }
+
+ #[test]
+ fn clamp_request_timeout_caps_an_oversized_value_at_the_ceiling() {
+ assert_eq!(clamp_request_timeout(i32::MAX), MAX_REQUEST_TIMEOUT);
+ }
+
+ #[test]
+ fn clamp_request_timeout_passes_through_a_reasonable_value_unchanged() {
+ assert_eq!(clamp_request_timeout(5_000), Duration::from_secs(5));
+ }
+}
diff --git a/gateways/kafka/tests/delete_topics_real_bridge_tests.rs b/gateways/kafka/tests/delete_topics_real_bridge_tests.rs
new file mode 100644
index 000000000..c6c592cbb
--- /dev/null
+++ b/gateways/kafka/tests/delete_topics_real_bridge_tests.rs
@@ -0,0 +1,322 @@
+// 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.
+
+//! Wire-level `DeleteTopics` tests against a real `iggy-server` process, through
+//! [`delete_topics::handle`] with a connected [`GatewayState`] - same shape as
+//! `create_topics_real_bridge_tests.rs`. Exercises the handler's own orchestration (the
+//! topic-cap check, per-topic independent deletion, real error codes reaching the wire) and,
+//! critically, the deliberate choice this bridge makes not to delete the backing Iggy stream.
+//!
+//! Requests are hand-built at the v5 flexible wire shape, for the same reason
+//! `create_topics_real_bridge_tests.rs` hand-builds its own: this crate builds with
+//! `default-features = false, features = ["broker"]`, which gives request `Decodable` and
+//! response `Encodable` but not the reverse.
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use iggy::prelude::{Identifier, StreamClient, TopicClient};
+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_NONE, ERROR_POLICY_VIOLATION, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ GatewayState,
+};
+use iggy_gateway_kafka::protocol::handlers::delete_topics;
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+
+use codec::{Decoder, Encoder};
+use iggy_server::{TestServer, raw_client};
+
+const REQUEST_VERSION: i16 = 5;
+const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+
+/// Builds a v5 flexible `DeleteTopics` request body for one or more topic names.
+fn build_request(names: &[&str]) -> Bytes {
+ let mut enc = Encoder::with_capacity(128);
+ enc.write_varint((names.len() + 1) as u64);
+ for name in names {
+ enc.write_compact_nullable_string(Some(name));
+ }
+ enc.write_i32(5_000); // timeout_ms
+ enc.write_empty_tagged_fields();
+ enc.freeze()
+}
+
+/// Decodes every topic result in a v5 flexible `DeleteTopics` response into `(name, error_code)`,
+/// in wire order.
+fn decode_all_results(body: Bytes) -> Vec<(Option<String>, i16)> {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let results_plus_one = d.read_varint().expect("responses array count");
+ let mut results = Vec::new();
+ for _ in 1..results_plus_one {
+ let name = d.read_compact_nullable_string().expect("topic name");
+ let error_code = d.read_i16().expect("error_code");
+ let _error_message = d.read_compact_nullable_string().expect("error_message");
+ let _tagged = d.read_varint().expect("topic tagged fields");
+ results.push((name, error_code));
+ }
+ results
+}
+
+/// Decodes the first (only) topic result into `(error_code, error_message)`.
+fn decode_first_result(body: Bytes) -> (i16, Option<String>) {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let _results_plus_one = d.read_varint().expect("responses array count");
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ let error_code = d.read_i16().expect("error_code");
+ let error_message = d.read_compact_nullable_string().expect("error_message");
+ (error_code, error_message)
+}
+
+async fn send(state: &GatewayState, names: &[&str]) -> Vec<(Option<String>, i16)> {
+ let body = build_request(names);
+ let outcome = delete_topics::handle(state, REQUEST_VERSION, body).await;
+ let resp_body = outcome.expect_response("DeleteTopics request always answers");
+ decode_all_results(resp_body)
+}
+
+async fn connected_state(server: &TestServer) -> GatewayState {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ TEST_MAX_FRAME_SIZE,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+}
+
+#[tokio::test]
+#[serial]
+async fn delete_topics_deletes_a_real_iggy_topic() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ let Some(bridge) = &state.bridge else {
+ panic!("bridge must be connected")
+ };
+ bridge
+ .create_kafka_topic("orders", 3)
+ .await
+ .expect("create the topic to delete");
+
+ let results = send(&state, &["orders"]).await;
+ assert_eq!(results, vec![(Some("orders".to_string()), ERROR_NONE)]);
+
+ let raw = raw_client(&server).await;
+ let topics = raw
+ .get_topics(&Identifier::named("kafka").expect("valid stream name"))
+ .await
+ .expect("get_topics call");
+ assert!(
+ topics.is_empty(),
+ "the handler must have actually deleted the topic"
+ );
+}
+
+/// The design decision this PR made deliberately (not the issue's literal "delete stream +
+/// topic" wording): `DeleteTopics` must never delete the backing Iggy stream, since the shared
+/// `default_stream` can hold other Kafka topics' data this bridge has no way to distinguish from
+/// an abandoned one. Verified here by creating two topics in the same (default) stream, deleting
+/// one, and confirming the stream - and the other topic in it - are both still there.
+#[tokio::test]
+#[serial]
+async fn delete_topics_never_deletes_the_backing_stream_even_when_it_becomes_empty() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ let Some(bridge) = &state.bridge else {
+ panic!("bridge must be connected")
+ };
+ bridge
+ .create_kafka_topic("orders", 1)
+ .await
+ .expect("create orders");
+ bridge
+ .create_kafka_topic("payments", 1)
+ .await
+ .expect("create payments");
+
+ // Delete both topics sharing the default stream - if this ever deleted the stream once it
+ // emptied out, the second delete (or the stream-existence check below) would fail.
+ let results = send(&state, &["orders", "payments"]).await;
+ assert_eq!(
+ results,
+ vec![
+ (Some("orders".to_string()), ERROR_NONE),
+ (Some("payments".to_string()), ERROR_NONE),
+ ]
+ );
+
+ let raw = raw_client(&server).await;
+ let streams = raw.get_streams().await.expect("get_streams call");
+ assert_eq!(
+ streams.len(),
+ 1,
+ "the now-empty default stream must still exist - DeleteTopics never touches it"
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn delete_topics_on_a_nonexistent_topic_returns_unknown_topic_or_partition() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+
+ let results = send(&state, &["never-existed"]).await;
+ assert_eq!(
+ results,
+ vec![(
+ Some("never-existed".to_string()),
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION
+ )]
+ );
+}
+
+/// A name repeated in the same request is not specially rejected the way `CreateTopics` rejects
+/// a duplicate - the first occurrence deletes the topic, so the second genuinely finds nothing.
+#[tokio::test]
+#[serial]
+async fn delete_topics_second_occurrence_of_a_duplicate_name_is_unknown_topic_or_partition() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ let Some(bridge) = &state.bridge else {
+ panic!("bridge must be connected")
+ };
+ bridge
+ .create_kafka_topic("orders", 1)
+ .await
+ .expect("create orders");
+
+ let results = send(&state, &["orders", "orders"]).await;
+ assert_eq!(
+ results,
+ vec![
+ (Some("orders".to_string()), ERROR_NONE),
+ (Some("orders".to_string()), ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ ]
+ );
+}
+
+/// One missing topic in a batch must not block an unrelated, existing topic from being deleted.
+#[tokio::test]
+#[serial]
+async fn delete_topics_deletes_existing_topics_alongside_a_missing_one() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ let Some(bridge) = &state.bridge else {
+ panic!("bridge must be connected")
+ };
+ bridge
+ .create_kafka_topic("orders", 1)
+ .await
+ .expect("create orders");
+
+ let results = send(&state, &["orders", "never-existed"]).await;
+ assert_eq!(
+ results,
+ vec![
+ (Some("orders".to_string()), ERROR_NONE),
+ (
+ Some("never-existed".to_string()),
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION
+ ),
+ ]
+ );
+
+ let raw = raw_client(&server).await;
+ let topics = raw
+ .get_topics(&Identifier::named("kafka").expect("valid stream name"))
+ .await
+ .expect("get_topics call");
+ assert!(topics.is_empty(), "orders must have actually been deleted");
+}
+
+/// Regression test: a request naming more than the bridge-backed topic cap must be rejected
+/// wholesale (every entry, `POLICY_VIOLATION`, nothing deleted) rather than partially served -
+/// same contract `create_topics_rejects_more_than_the_topic_cap` holds for `CreateTopics`.
+#[tokio::test]
+#[serial]
+async fn delete_topics_rejects_more_than_the_topic_cap() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ let Some(bridge) = &state.bridge else {
+ panic!("bridge must be connected")
+ };
+ bridge
+ .create_kafka_topic("orders", 1)
+ .await
+ .expect("create orders");
+
+ let names: Vec<String> = (0..101).map(|i| format!("topic-{i}")).collect();
+ let name_refs: Vec<&str> = names.iter().map(String::as_str).collect();
+ let results = send(&state, &name_refs).await;
+ assert_eq!(results.len(), 101);
+ for (_, error_code) in &results {
+ assert_eq!(*error_code, ERROR_POLICY_VIOLATION);
+ }
+
+ let raw = raw_client(&server).await;
+ let topics = raw
+ .get_topics(&Identifier::named("kafka").expect("valid stream name"))
+ .await
+ .expect("get_topics call");
+ assert_eq!(
+ topics.len(),
+ 1,
+ "an over-cap request must delete nothing, including topics not named in it"
+ );
+}
+
+/// `error_message` carries the real cause for a client-visible rejection path (the topic-cap
+/// policy violation), not just a bare code.
+#[tokio::test]
+#[serial]
+async fn delete_topics_over_cap_error_message_names_the_limit() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+
+ let names: Vec<String> = (0..101).map(|i| format!("topic-{i}")).collect();
+ let name_refs: Vec<&str> = names.iter().map(String::as_str).collect();
+ let body = build_request(&name_refs);
+ let outcome = delete_topics::handle(&state, REQUEST_VERSION, body).await;
+ let resp_body = outcome.expect_response("DeleteTopics request always answers");
+ let (error_code, error_message) = decode_first_result(resp_body);
+ assert_eq!(error_code, ERROR_POLICY_VIOLATION);
+ assert!(
+ error_message.is_some_and(|m| m.contains("100")),
+ "error_message must name the actual cap"
+ );
+}