This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a commit to branch feat/gateways-3546-delete-topics-only
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to
refs/heads/feat/gateways-3546-delete-topics-only by this push:
new 657228341 Fixing review comments
657228341 is described below
commit 657228341ea6394ddc083ab545083ccc1307fd34
Author: ryerraguntla <[email protected]>
AuthorDate: Wed Oct 7 06:27:44 2026 -0400
Fixing review comments
---
gateways/kafka/README.md | 5 +-
gateways/kafka/docs/SCOPE.md | 18 +++-
gateways/kafka/src/protocol/api.rs | 9 +-
.../kafka/src/protocol/handlers/create_topics.rs | 79 +++++++++-----
.../kafka/src/protocol/handlers/delete_topics.rs | 115 +++++++++++++++++++--
gateways/kafka/src/protocol/handlers/mod.rs | 4 +-
gateways/kafka/src/server.rs | 3 +
gateways/kafka/tests/api_handler_tests.rs | 58 +++++++++--
gateways/kafka/tests/common/wire.rs | 23 +++++
.../kafka/tests/delete_topics_real_bridge_tests.rs | 32 ++++--
gateways/kafka/tests/fetch_real_bridge_tests.rs | 10 +-
11 files changed, 293 insertions(+), 63 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 0145462cc..f95f599da 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -9,7 +9,10 @@ Foundation layer for
[apache/iggy#3421](https://github.com/apache/iggy/issues/34
> **CreateTopics and DeleteTopics run as the bridge's own Iggy user**: with
> the bridge on and
> `IGGY_KAFKA_SASL_ENABLED` off (the default), any client that can reach this
> port can create
> topics (up to 1000 partitions each) and can delete topics and their data.
> Every client reads and
-> writes as the bridge's Iggy user. See [docs/SCOPE.md](docs/SCOPE.md).
+> writes as the bridge's Iggy user. With SASL on, `DeleteTopics` additionally
requires the
+> authenticated principal to hold `manage_topics`; without it every name in
the request fails
+> `TOPIC_AUTHORIZATION_FAILED` (29) and nothing is deleted. `CreateTopics` has
no such gate yet -
+> SASL on does not protect it. See [docs/SCOPE.md](docs/SCOPE.md).
>
> InitProducerId does real work too, with or without the bridge: it allocates
> a producer id, so a stock idempotent producer starts instead of failing at
> startup.
>
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 09d5deafe..234a7525f 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -243,13 +243,21 @@ below it are still open for the issues that build on top
of it.
[`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md).
- Per-topic results are independent: one missing or invalid name in a batch
does not block an
existing, validly-named topic elsewhere in the same request from being
deleted.
- - A name repeated within one 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, so the first occurrence deletes it and the second then
correctly answers
- `UNKNOWN_TOPIC_OR_PARTITION` (3) because the topic genuinely no longer
exists.
+ - A name repeated within one request is rejected outright for every
occurrence
+ (`INVALID_REQUEST`, 42, no bridge call for any occurrence), the same
choice `CreateTopics`
+ makes for a repeated topic name: real Kafka's
`ControllerApis.deleteTopics` answers
+ `INVALID_REQUEST` "Duplicate topic name" for every occurrence and deletes
nothing, rather
+ than a first-occurrence-wins split that would let an `AdminClient` caller
observe a delete it
+ never got a clean per-name answer for.
- Same bridge-fan-out bounds as CreateTopics: a request naming more than 100
topic names is
rejected outright (`POLICY_VIOLATION`, no bridge call for any of them),
and the wire
- `timeout_ms` (clamped to `[1s, 30s]`) bounds the whole handler's aggregate
bridge work.
+ `timeout_ms` (clamped to `[1s, 30s]`) bounds the whole handler's aggregate
bridge work. The
+ deadline is checked before each name's own bridge call starts, not only
inside the
+ `timeout_at` wrapping it - `timeout_at` polls its inner future before it
checks the delay, so
+ a deadline that already passed would otherwise still start a real delete.
+ - With SASL on, the authenticated principal must hold `manage_topics` -
without it every name
+ in the request fails `TOPIC_AUTHORIZATION_FAILED` (29) and
`delete_kafka_topic` is never
+ called. With SASL off there is no principal to check, and the request is
not gated here.
- v6 (the `topics: Vec<DeleteTopicState>`, topic-id-based shape) is not
advertised - this bridge
has no concept of a Kafka topic id, only the Kafka-side name
`TopicMapping` resolves.
- [x] Document partition mapping in [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md):
diff --git a/gateways/kafka/src/protocol/api.rs
b/gateways/kafka/src/protocol/api.rs
index f8965a22b..b65916a44 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -26,6 +26,7 @@ use kafka_protocol::messages::{
use tokio::sync::Semaphore;
use tokio_util::sync::CancellationToken;
+use crate::auth::AuthenticatedPrincipal;
use crate::bridge::IggyBridge;
use crate::error::Result;
use crate::group::{GroupCoordinator, GroupCoordinatorConfig};
@@ -407,18 +408,22 @@ pub async fn handle_request_bounded(
body: Bytes,
) -> HandleOutcome {
let connection = ConnectionState::default();
- handle_connection_request(state, &connection, api_key, api_version,
body).await
+ handle_connection_request(state, &connection, None, api_key, api_version,
body).await
}
/// [`handle_request_bounded`] for one request of `connection`.
+///
+/// `principal` is `None` when SASL is off, or for a key the connection
reached before
+/// authenticating - only [`crate::server::route_frame`] ever has a principal
to pass.
pub async fn handle_connection_request(
state: &GatewayState,
connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
api_key: i16,
api_version: i16,
body: Bytes,
) -> HandleOutcome {
- dispatch(state, connection, api_key, api_version, body).await
+ dispatch(state, connection, principal, api_key, api_version, body).await
}
#[must_use]
diff --git a/gateways/kafka/src/protocol/handlers/create_topics.rs
b/gateways/kafka/src/protocol/handlers/create_topics.rs
index b411d5f4e..69a38fbf1 100644
--- a/gateways/kafka/src/protocol/handlers/create_topics.rs
+++ b/gateways/kafka/src/protocol/handlers/create_topics.rs
@@ -168,9 +168,13 @@ pub async fn handle(state: &GatewayState, api_version:
i16, body: Bytes) -> Hand
/// `deadline` bounds each topic's own bridge work individually
(`timeout_at`), not the whole
/// loop: a single `timeout` around the entire call would discard every
already-resolved result
/// the moment one topic's call ran long, answering `REQUEST_TIMED_OUT` even
for topics that had
-/// already committed. Once `deadline` passes, every remaining topic's own
`timeout_at` elapses
-/// immediately rather than making a fresh bridge call, so a stuck topic near
the front of a large
-/// batch does not turn into one slow round trip per topic behind it.
+/// already committed.
+///
+/// `tokio::time::timeout_at` polls the inner future before it ever checks the
deadline (see
+/// `Timeout::poll`), so once `deadline` has already passed, wrapping a fresh
bridge call in it
+/// still starts that call - a topic past the deadline would both answer
`REQUEST_TIMED_OUT` *and*
+/// actually get created. The `Instant::now() >= deadline` check below must
run and must skip the
+/// call itself, the same pattern `list_offsets.rs` uses, not rely on
`timeout_at` to skip it.
async fn create_all_topics(
bridge: &IggyBridge,
api_version: i16,
@@ -180,31 +184,52 @@ async fn create_all_topics(
deadline: Instant,
) -> Vec<CreatableTopicResult> {
let mut results = Vec::with_capacity(topics.len());
+ let mut deadline_exceeded = false;
for topic in topics {
- let result = if duplicate_names.contains(&topic.name) {
- CreatableTopicResult::default()
- .with_name(topic.name.clone())
- .with_error_code(ERROR_INVALID_REQUEST)
- .with_error_message(None)
- } else {
- match tokio::time::timeout_at(
- deadline,
- create_one_topic(bridge, api_version, topic, validate_only),
- )
- .await
- {
- Ok(result) => result,
- Err(_elapsed) => {
- tracing::warn!(
- kafka_topic = topic.name.as_str(),
- "CreateTopics: this topic's bridge work exceeded the
request deadline; \
- answering retriable instead of blocking further"
- );
- CreatableTopicResult::default()
- .with_name(topic.name.clone())
- .with_error_code(ERROR_REQUEST_TIMED_OUT)
- .with_error_message(None)
- }
+ if duplicate_names.contains(&topic.name) {
+ results.push(
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_INVALID_REQUEST)
+ .with_error_message(None),
+ );
+ continue;
+ }
+ if deadline_exceeded || Instant::now() >= deadline {
+ if !deadline_exceeded {
+ deadline_exceeded = true;
+ tracing::warn!(
+ "CreateTopics deadline passed; answering remaining topics
retriable \
+ instead of starting new Iggy calls"
+ );
+ }
+ results.push(
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_REQUEST_TIMED_OUT)
+ .with_error_message(None),
+ );
+ continue;
+ }
+
+ let result = match tokio::time::timeout_at(
+ deadline,
+ create_one_topic(bridge, api_version, topic, validate_only),
+ )
+ .await
+ {
+ Ok(result) => result,
+ Err(_elapsed) => {
+ deadline_exceeded = true;
+ tracing::warn!(
+ kafka_topic = topic.name.as_str(),
+ "CreateTopics: this topic's bridge work exceeded the
request deadline; \
+ answering retriable instead of blocking further"
+ );
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_REQUEST_TIMED_OUT)
+ .with_error_message(None)
}
};
results.push(result);
diff --git a/gateways/kafka/src/protocol/handlers/delete_topics.rs
b/gateways/kafka/src/protocol/handlers/delete_topics.rs
index 319361591..ef776ae4b 100644
--- a/gateways/kafka/src/protocol/handlers/delete_topics.rs
+++ b/gateways/kafka/src/protocol/handlers/delete_topics.rs
@@ -27,6 +27,8 @@
//! silently dropped. Unlike `CreateTopics`, whose equivalent field is encoded
at every version it
//! supports.
+use std::collections::HashSet;
+
use bytes::Bytes;
use iggy::prelude::IggyError;
use kafka_protocol::messages::delete_topics_response::DeletableTopicResult;
@@ -35,12 +37,13 @@ use kafka_protocol::protocol::StrBytes;
use tokio::time::Instant;
+use crate::auth::AuthenticatedPrincipal;
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,
+ ERROR_NOT_CONTROLLER, ERROR_POLICY_VIOLATION, ERROR_REQUEST_TIMED_OUT,
+ ERROR_TOPIC_AUTHORIZATION_FAILED, GatewayState, HandleOutcome,
};
use crate::protocol::bounds_guard::validate_delete_topics_shape;
use crate::protocol::handlers::{
@@ -66,7 +69,12 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
/// rather than the distinct one `CreateTopics` charges.
const MAX_BRIDGE_BACKED_TOPICS: usize = 100;
-pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) ->
HandleOutcome {
+pub async fn handle(
+ state: &GatewayState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
let Some(bridge) = &state.bridge else {
return handle_versioned_request(
API_KEY_DELETE_TOPICS,
@@ -103,6 +111,11 @@ pub async fn handle(state: &GatewayState, api_version:
i16, body: Bytes) -> Hand
}
};
+ if let Some(results) = authorize(principal, &req.topic_names) {
+ let resp = DeleteTopicsResponse::default().with_responses(results);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"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 {
@@ -128,12 +141,66 @@ pub async fn handle(state: &GatewayState, api_version:
i16, body: Bytes) -> Hand
return respond_or_close(encode_message(&resp, api_version, 256),
"DeleteTopics");
}
+ // Real Kafka (`ControllerApis.deleteTopics`) refuses every occurrence of
a duplicate name
+ // with `INVALID_REQUEST` (42) and deletes nothing for it, the same choice
this gateway's own
+ // `CreateTopics` already makes for a repeated topic name.
+ let duplicate_names = find_duplicate_names(&req.topic_names);
+
let deadline = Instant::now() + clamp_request_timeout(req.timeout_ms);
- let results = delete_all_topics(bridge, &req.topic_names, deadline).await;
+ let results = delete_all_topics(bridge, &req.topic_names,
&duplicate_names, deadline).await;
let resp = DeleteTopicsResponse::default().with_responses(results);
respond_or_close(encode_message(&resp, api_version, 256), "DeleteTopics")
}
+/// Denies every requested topic when `principal` is known but is not allowed
to delete topic
+/// data, or when its permissions could not be read. `None` means SASL is off,
which this
+/// gateway treats as no principal to enforce against - the same choice the
rest of this
+/// gateway makes when authentication itself is not configured.
+fn authorize(
+ principal: Option<&AuthenticatedPrincipal>,
+ topic_names: &[TopicName],
+) -> Option<Vec<DeletableTopicResult>> {
+ let principal = principal?;
+ if principal.permissions_known && principal.permissions.manage_topics {
+ return None;
+ }
+ if !principal.permissions_known {
+ // The permission read failed after a successful login, so this
connection holds no real
+ // answer. Denying is the fail-closed choice: granting would authorize
an irreversible
+ // delete off a value nothing ever actually read.
+ tracing::warn!(
+ principal = %principal.username,
+ "DeleteTopics asked on a connection whose permissions were never
read; denying"
+ );
+ }
+ let message =
+ StrBytes::from_static_str("the authenticated principal is not
authorized to delete topics");
+ Some(
+ topic_names
+ .iter()
+ .map(|name| {
+ DeletableTopicResult::default()
+ .with_name(Some(name.clone()))
+ .with_error_code(ERROR_TOPIC_AUTHORIZATION_FAILED)
+ .with_error_message(Some(message.clone()))
+ })
+ .collect(),
+ )
+}
+
+/// Every topic name that appears more than once in `topic_names` - mirrors
+/// `create_topics::find_duplicate_names`.
+fn find_duplicate_names(topic_names: &[TopicName]) -> HashSet<TopicName> {
+ let mut seen = HashSet::with_capacity(topic_names.len());
+ let mut duplicates = HashSet::new();
+ for name in topic_names {
+ if !seen.insert(name.clone()) {
+ duplicates.insert(name.clone());
+ }
+ }
+ duplicates
+}
+
/// Deletes every requested topic, independently of the others.
///
/// `deadline` bounds each topic's own bridge work individually
(`timeout_at`), not the whole
@@ -141,21 +208,51 @@ pub async fn handle(state: &GatewayState, api_version:
i16, body: Bytes) -> Hand
/// 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.
+/// `tokio::time::timeout_at` polls the inner future before it ever checks the
deadline (see
+/// `Timeout::poll`), so once `deadline` has already passed, wrapping a fresh
bridge call in it
+/// still starts that call - a name past the deadline would both answer
`REQUEST_TIMED_OUT` *and*
+/// actually delete the topic. The `Instant::now() >= deadline` check below
must run and must
+/// skip the call itself, the same pattern `list_offsets.rs` uses, not rely on
`timeout_at` to
+/// skip it.
async fn delete_all_topics(
bridge: &IggyBridge,
topic_names: &[TopicName],
+ duplicate_names: &HashSet<TopicName>,
deadline: Instant,
) -> Vec<DeletableTopicResult> {
let mut results = Vec::with_capacity(topic_names.len());
+ let mut deadline_exceeded = false;
for name in topic_names {
+ if duplicate_names.contains(name) {
+ results.push(
+ DeletableTopicResult::default()
+ .with_name(Some(name.clone()))
+ .with_error_code(ERROR_INVALID_REQUEST)
+ .with_error_message(None),
+ );
+ continue;
+ }
+ if deadline_exceeded || Instant::now() >= deadline {
+ if !deadline_exceeded {
+ deadline_exceeded = true;
+ tracing::warn!(
+ "DeleteTopics deadline passed; answering remaining topics
retriable \
+ instead of starting new Iggy calls"
+ );
+ }
+ results.push(
+ DeletableTopicResult::default()
+ .with_name(Some(name.clone()))
+ .with_error_code(ERROR_REQUEST_TIMED_OUT)
+ .with_error_message(None),
+ );
+ continue;
+ }
+
let result = match tokio::time::timeout_at(deadline,
delete_one_topic(bridge, name)).await {
Ok(result) => result,
Err(_elapsed) => {
+ deadline_exceeded = true;
tracing::warn!(
kafka_topic = name.as_str(),
"DeleteTopics: this topic's bridge work exceeded the
request deadline; \
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs
b/gateways/kafka/src/protocol/handlers/mod.rs
index 89bfab1ef..dbc371975 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -45,6 +45,7 @@ use kafka_protocol::messages::TransactionalId;
use kafka_protocol::protocol::{Decodable, Encodable};
use tokio::runtime::{Handle, RuntimeFlavor};
+use crate::auth::AuthenticatedPrincipal;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DELETE_TOPICS,
API_KEY_FETCH,
@@ -92,6 +93,7 @@ pub(crate) fn off_worker<T>(heavy: bool, work: impl FnOnce()
-> T) -> T {
pub async fn dispatch(
state: &GatewayState,
connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
api_key: i16,
api_version: i16,
body: Bytes,
@@ -103,7 +105,7 @@ pub async fn dispatch(
API_KEY_METADATA => metadata::handle(state, api_version, body).await,
API_KEY_API_VERSIONS => api_versions::handle(state, api_version,
body).await,
API_KEY_CREATE_TOPICS => create_topics::handle(state, api_version,
body).await,
- API_KEY_DELETE_TOPICS => delete_topics::handle(state, api_version,
body).await,
+ API_KEY_DELETE_TOPICS => delete_topics::handle(state, principal,
api_version, body).await,
API_KEY_FIND_COORDINATOR => find_coordinator::handle(state,
api_version, body).await,
API_KEY_JOIN_GROUP => join_group::handle(state, api_version,
body).await,
API_KEY_HEARTBEAT => heartbeat::handle(state, api_version, body).await,
diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs
index 8315e73fa..70ee83e2e 100644
--- a/gateways/kafka/src/server.rs
+++ b/gateways/kafka/src/server.rs
@@ -550,6 +550,7 @@ async fn route_frame(
handle_connection_request(
ctx.state,
ctx.connection,
+ principal.as_ref(),
req.request_api_key,
req.request_api_version,
body,
@@ -563,9 +564,11 @@ async fn route_frame(
// peer repeat a rejected version forever, resetting the
pre-authentication read budget
// on every frame and holding a `max_connections` permit with it.
sasl_state.count_api_versions_answer();
+ // Pre-authentication by construction: no principal exists yet at
this point.
handle_connection_request(
ctx.state,
ctx.connection,
+ None,
req.request_api_key,
req.request_api_version,
body,
diff --git a/gateways/kafka/tests/api_handler_tests.rs
b/gateways/kafka/tests/api_handler_tests.rs
index d007221bb..eff5ff440 100644
--- a/gateways/kafka/tests/api_handler_tests.rs
+++ b/gateways/kafka/tests/api_handler_tests.rs
@@ -29,10 +29,10 @@ mod wire;
use bytes::Bytes;
use iggy_gateway_kafka::protocol::api::{
- API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_FETCH,
API_KEY_LIST_OFFSETS,
- API_KEY_METADATA, API_KEY_PRODUCE, ERROR_INVALID_REQUEST, ERROR_NONE,
ERROR_NOT_CONTROLLER,
- ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
ERROR_UNSUPPORTED_VERSION,
- handle_request, is_supported_version, supported_api_ranges,
+ API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DELETE_TOPICS,
API_KEY_FETCH,
+ API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_PRODUCE,
ERROR_INVALID_REQUEST, ERROR_NONE,
+ ERROR_NOT_CONTROLLER, ERROR_NOT_LEADER_OR_FOLLOWER,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ ERROR_UNSUPPORTED_VERSION, handle_request, is_supported_version,
supported_api_ranges,
};
use codec::Decoder;
@@ -40,9 +40,10 @@ use fixtures::load_fixture_body_or_skip;
use scope::default_broker;
use tcp::{build_metadata_legacy_request, build_produce_v3_body};
use wire::{
- build_api_versions_flexible_request, build_metadata_all_topics_legacy,
- build_metadata_flexible_request, build_metadata_flexible_request_v10,
- build_metadata_legacy_request_for_version, build_produce_legacy_request,
+ build_api_versions_flexible_request, build_delete_topics_request_with_name,
+ build_metadata_all_topics_legacy, build_metadata_flexible_request,
+ build_metadata_flexible_request_v10,
build_metadata_legacy_request_for_version,
+ build_produce_legacy_request,
};
// ── ApiVersions ─────────────────────────────────────────────────────────────
@@ -808,6 +809,49 @@ async fn
create_topics_decodes_request_with_real_topic_and_assignment() {
assert_eq!(d.read_i16().unwrap(), ERROR_NOT_CONTROLLER);
}
+/// Pins the `flexible = version >= 4` boundary `validate_delete_topics_shape`
and
+/// `encode_delete_topics_response` both rely on, with a real (non-empty)
topic name so the
+/// per-topic error code is actually observable in the response - an empty
request's response
+/// has zero entries, so no error code in it is ever asserted anywhere else.
v1 and v3 are
+/// legacy; v4 and v5 are flexible (v4 has no `error_message` field, v5 does).
+#[tokio::test]
+async fn
delete_topics_stub_response_returns_not_controller_per_version_shape() {
+ for version in [1i16, 3, 4, 5] {
+ let body = build_delete_topics_request_with_name(version, "orders");
+ let resp = handle_request(API_KEY_DELETE_TOPICS, version, body,
&default_broker())
+ .await
+ .expect_response("test request has acks != 0 and expects a
response");
+ let flexible = version >= 4;
+ let mut d = Decoder::new(resp);
+ let _throttle = d.read_i32().unwrap();
+ if flexible {
+ let responses_plus_one = d.read_varint().unwrap();
+ assert_eq!(
+ responses_plus_one, 2,
+ "DeleteTopics v{version}: one response"
+ );
+ assert_eq!(
+ d.read_compact_nullable_string().unwrap(),
+ Some("orders".to_string()),
+ "DeleteTopics v{version}: name must echo the request"
+ );
+ } else {
+ let responses = d.read_i32().unwrap();
+ assert_eq!(responses, 1, "DeleteTopics v{version}: one response");
+ assert_eq!(
+ d.read_nullable_string().unwrap(),
+ Some("orders".to_string()),
+ "DeleteTopics v{version}: name must echo the request"
+ );
+ }
+ assert_eq!(
+ d.read_i16().unwrap(),
+ ERROR_NOT_CONTROLLER,
+ "DeleteTopics v{version}"
+ );
+ }
+}
+
// ── Produce acks=0 with a malformed body closes ──────
#[tokio::test]
diff --git a/gateways/kafka/tests/common/wire.rs
b/gateways/kafka/tests/common/wire.rs
index e85d17553..bf16efaaa 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -195,6 +195,29 @@ pub fn build_delete_topics_empty_request(version: i16) ->
Bytes {
enc.freeze()
}
+/// `DeleteTopics` request naming one topic, for supported versions (v1-v5).
+///
+/// Unlike [`build_delete_topics_empty_request`], this actually produces a
per-topic result -
+/// an empty request's response has zero entries, so no error code in it is
ever observed.
+pub fn build_delete_topics_request_with_name(version: i16, name: &str) ->
Bytes {
+ let flexible = version >= 4;
+ let mut enc = Encoder::with_capacity(32);
+
+ if flexible {
+ enc.write_varint(2); // one topic_names entry (N+1 = 2)
+ enc.write_compact_nullable_string(Some(name));
+ } else {
+ enc.write_i32(1);
+ enc.write_nullable_string(Some(name)).expect("name fits");
+ }
+ enc.write_i32(5_000); // timeout_ms
+ if flexible {
+ enc.write_empty_tagged_fields();
+ }
+
+ enc.freeze()
+}
+
/// Produce v2–v8 legacy request with optional transactional id and topic.
pub fn build_produce_legacy_request(
version: i16,
diff --git a/gateways/kafka/tests/delete_topics_real_bridge_tests.rs
b/gateways/kafka/tests/delete_topics_real_bridge_tests.rs
index c6c592cbb..f125d3f2a 100644
--- a/gateways/kafka/tests/delete_topics_real_bridge_tests.rs
+++ b/gateways/kafka/tests/delete_topics_real_bridge_tests.rs
@@ -36,8 +36,8 @@ 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,
+ BrokerAdvertise, ERROR_INVALID_REQUEST, ERROR_NONE, ERROR_POLICY_VIOLATION,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState,
};
use iggy_gateway_kafka::protocol::handlers::delete_topics;
@@ -94,7 +94,7 @@ fn decode_first_result(body: Bytes) -> (i16, Option<String>) {
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 outcome = delete_topics::handle(state, None, REQUEST_VERSION,
body).await;
let resp_body = outcome.expect_response("DeleteTopics request always
answers");
decode_all_results(resp_body)
}
@@ -201,11 +201,15 @@ async fn
delete_topics_on_a_nonexistent_topic_returns_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.
+/// A name repeated in the same request is rejected outright for every
occurrence, the same
+/// choice `CreateTopics` makes for a repeated topic name (real Kafka's
`ControllerApis.deleteTopics`
+/// answers `INVALID_REQUEST`(42) "Duplicate topic name" for every occurrence
and deletes nothing -
+/// a first-occurrence-wins split would let an `AdminClient` caller observe a
delete it never got
+/// a clean answer for, since the client keys its futures by name and silently
discards the second
+/// per-name result regardless of which occurrence this bridge picked).
#[tokio::test]
#[serial]
-async fn
delete_topics_second_occurrence_of_a_duplicate_name_is_unknown_topic_or_partition()
{
+async fn delete_topics_every_occurrence_of_a_duplicate_name_is_rejected() {
let data_dir = tempfile::tempdir().expect("tempdir");
let server = TestServer::spawn(data_dir.path()).await;
let state = connected_state(&server).await;
@@ -221,10 +225,20 @@ async fn
delete_topics_second_occurrence_of_a_duplicate_name_is_unknown_topic_or
assert_eq!(
results,
vec![
- (Some("orders".to_string()), ERROR_NONE),
- (Some("orders".to_string()), ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ (Some("orders".to_string()), ERROR_INVALID_REQUEST),
+ (Some("orders".to_string()), ERROR_INVALID_REQUEST),
]
);
+
+ // Neither occurrence deleted anything - the topic must still exist.
+ let topic = bridge
+ .get_kafka_topic("orders")
+ .await
+ .expect("lookup should not fail");
+ assert!(
+ topic.is_some(),
+ "a duplicate name must not delete the topic it names"
+ );
}
/// One missing topic in a batch must not block an unrelated, existing topic
from being deleted.
@@ -311,7 +325,7 @@ async fn
delete_topics_over_cap_error_message_names_the_limit() {
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 outcome = delete_topics::handle(&state, None, 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);
diff --git a/gateways/kafka/tests/fetch_real_bridge_tests.rs
b/gateways/kafka/tests/fetch_real_bridge_tests.rs
index 1340aedad..ecf480ff0 100644
--- a/gateways/kafka/tests/fetch_real_bridge_tests.rs
+++ b/gateways/kafka/tests/fetch_real_bridge_tests.rs
@@ -249,8 +249,14 @@ async fn fetch_on(
request
.encode(&mut body, VERSION)
.expect("the request encodes at this version");
- let handled =
- handle_connection_request(state, connection, API_KEY_FETCH, VERSION,
body.freeze());
+ let handled = handle_connection_request(
+ state,
+ connection,
+ None,
+ API_KEY_FETCH,
+ VERSION,
+ body.freeze(),
+ );
let mut response = tokio::time::timeout(PROMPT, handled)
.await
.expect("records or an error answer at once")
