This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 4dec2a985 feat(gateways): Kafka Gateway add DeleteTopics API Key that
deletes the Iggy topic only (#4400)
4dec2a985 is described below
commit 4dec2a985528843bed7ba0a5f2c13137a45ffd99
Author: ryerraguntla <[email protected]>
AuthorDate: Mon Oct 12 07:46:49 2026 -0400
feat(gateways): Kafka Gateway add DeleteTopics API Key that deletes the
Iggy topic only (#4400)
---
.config/nextest.toml | 1 +
gateways/kafka/README.md | 23 +-
gateways/kafka/docs/BRIDGE_MAPPING.md | 21 ++
gateways/kafka/docs/MANUAL_TESTING.md | 10 +-
gateways/kafka/docs/SCOPE.md | 50 ++-
gateways/kafka/scripts/ci-wire-fixtures.sh | 2 +-
gateways/kafka/src/bridge/iggy_bridge/topics.rs | 25 ++
gateways/kafka/src/protocol/api.rs | 12 +-
gateways/kafka/src/protocol/bounds_guard.rs | 73 ++++
.../kafka/src/protocol/handlers/create_topics.rs | 125 ++++---
.../kafka/src/protocol/handlers/delete_topics.rs | 404 +++++++++++++++++++++
gateways/kafka/src/protocol/handlers/mod.rs | 47 ++-
gateways/kafka/tests/api_handler_tests.rs | 58 ++-
gateways/kafka/tests/common/scope.rs | 1 +
gateways/kafka/tests/common/wire.rs | 47 ++-
.../kafka/tests/delete_topics_real_bridge_tests.rs | 336 +++++++++++++++++
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 14 +-
gateways/kafka/tests/kafka_client_e2e_tests.rs | 71 ++++
gateways/kafka/tests/sasl_tests.rs | 6 +-
gateways/kafka/tests/version_firewall_tests.rs | 34 +-
20 files changed, 1235 insertions(+), 125 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index 23119ec0d..bd0f105e5 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -64,6 +64,7 @@ filter = '''
+ binary_id(iggy-gateway-kafka::list_offsets_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::produce_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::create_topics_real_bridge_tests)
+ + binary_id(iggy-gateway-kafka::delete_topics_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::metadata_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::offsets_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::phase1_e2e_tests)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index f78f6e155..6c729ce7d 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -1,18 +1,21 @@
# Kafka gateway (`iggy-gateway-kafka`)
-Foundation layer for
[apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener
on the Kafka wire port that decodes requests and validates scoped API keys and
versions. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics,
DescribeConfigs and AlterConfigs use Iggy. InitProducerId and consumer group
coordination work with or without one.
+Foundation layer for
[apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener
on the Kafka wire port that decodes requests and validates scoped API keys and
versions. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics,
DeleteTopics, DescribeConfigs and AlterConfigs use Iggy. InitProducerId and
consumer group coordination work with or without one.
> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch,
> ListOffsets, Metadata,
-> CreateTopics, DescribeConfigs and AlterConfigs use Iggy. With the bridge off
(the default), those
-> seven are stubs: Produce, Fetch and ListOffsets answer retriable
`NOT_LEADER_OR_FOLLOWER` (6),
-> CreateTopics, DescribeConfigs and AlterConfigs answer `NOT_CONTROLLER` (41),
and Metadata reports
-> every requested topic unknown. **CreateTopics and AlterConfigs 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) or
shorten a topic's
+> CreateTopics, DeleteTopics, DescribeConfigs and AlterConfigs use Iggy. With
the bridge off (the
+> default), those eight are stubs: Produce, Fetch and ListOffsets answer
retriable
+> `NOT_LEADER_OR_FOLLOWER` (6), CreateTopics, DeleteTopics, DescribeConfigs
and AlterConfigs answer
+> `NOT_CONTROLLER` (41), and Metadata reports every requested topic unknown.
**CreateTopics,
+> DeleteTopics and AlterConfigs 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), delete topics and their data, or
shorten a topic's
> `retention.ms` low enough that the cleaner drops sealed segments, deleting
> data. With SASL on,
-> `AlterConfigs` additionally requires the authenticated principal to hold
`manage_topics`; without
-> it every resource in the request fails `TOPIC_AUTHORIZATION_FAILED` (29) and
nothing is written.
-> Every client reads and writes as the bridge's Iggy user. See
[docs/SCOPE.md](docs/SCOPE.md).
+> `DeleteTopics` and `AlterConfigs` additionally require the authenticated
principal to hold
+> `manage_topics`; without it every resource/name in the request fails
+> `TOPIC_AUTHORIZATION_FAILED` (29) and nothing is written or deleted.
`CreateTopics` has no such
+> gate yet - SASL on does not protect it. Every client reads and writes as the
bridge's Iggy user.
+> 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/BRIDGE_MAPPING.md
b/gateways/kafka/docs/BRIDGE_MAPPING.md
index 2d5d2ecd1..0e025399d 100644
--- a/gateways/kafka/docs/BRIDGE_MAPPING.md
+++ b/gateways/kafka/docs/BRIDGE_MAPPING.md
@@ -347,6 +347,27 @@ to group members the way Kafka does, in the client, and
polls every partition by
Iggy's group registry is used as an offset key and for nothing else, which
[`OFFSET_STORAGE.md`](OFFSET_STORAGE.md) covers.
+## Topic deletion and the shared default stream
+
+`TopicMapping::resolve` sends every Kafka topic with no configured override to
the same
+`default_stream` - one Iggy stream holding many Iggy topics, one per Kafka
topic name. An
+override can point a specific Kafka topic at a different stream, but nothing
stops two different
+overrides, or an override and the default, from naming the same stream for two
different Kafka
+topics.
+
+`DeleteTopics` ([#3546](https://github.com/apache/iggy/issues/3546)) therefore
deletes the Iggy
+*topic* only, never the Iggy *stream* it lives in - even when that delete
leaves the stream with
+no topics left in it. A stream `DeleteTopics` happens to empty out is not the
same thing as a
+stream nothing will ever use again: the next `CreateTopics` for an unrelated
Kafka topic sharing
+that same default stream recreates exactly the state the delete just left, and
this bridge has no
+way to tell "abandoned" apart from "temporarily empty" from here. Real Kafka
has no analogous
+container object a topic delete would need to clean up either, so leaving the
stream alone also
+matches what a Kafka client actually expects to observe.
+
+`ensure_stream_and_topic`'s own doc comment
(`src/bridge/iggy_bridge/topics.rs`) flags the
+identical sharing risk from the create side: it never rolls back a stream it
just created on a
+failed topic create, for the same reason.
+
## Reserved header namespace
`kafka.` is reserved on messages the gateway writes. Nothing in the server
enforces it, so the
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md
b/gateways/kafka/docs/MANUAL_TESTING.md
index c323c0d34..072eea17a 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -216,6 +216,7 @@ For each API key, test **min−1**, **min**, **max**,
**max+1** using `kafka-mes
| 1 | Fetch | 4 | 12 | 3, 4, 12, 13 |
| 2 | ListOffsets | 1 | 6 | 0, 1, 6, 7 |
| 19 | CreateTopics | 2 | 5 | 1, 2, 5, 6 |
+| 20 | DeleteTopics | 1 | 5 | 0, 1, 5, 6 |
| 8 | OffsetCommit | 2 | 9 | 1, 2, 9, 10 |
| 9 | OffsetFetch | 1 | 9 | 0, 1, 9, 10 |
| 10 | FindCoordinator | 0 | 4 | −1, 0, 4, 5 |
@@ -231,8 +232,8 @@ For each API key, test **min−1**, **min**, **max**,
**max+1** using `kafka-mes
| ---- | ------ | ---------------------- | --------------------------- |
| B1 | ApiVersions negotiation | `error_code=0`; body lists 14 API keys with
correct min/max | KIP-511 exception: still answers, `error_code=35`
(UNSUPPORTED_VERSION), v0 response header regardless of the request's own
encoding |
| B2 | Metadata out-of-range | N/A | **Connection closes**, no response sent -
Metadata has no top-level error field to carry a version-correct error in |
-| B3 |
Produce/Fetch/ListOffsets/CreateTopics/InitProducerId/OffsetCommit/OffsetFetch
out-of-range | N/A | **Connection closes** for both above-max and below-min -
`kafka_protocol`'s schema floor for each of these five messages equals
`SUPPORTED_RANGES`' own min, so there is no encodable error response below min
either (see `SCOPE.md`'s Governance model) |
-| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys
0,1,2,3,8,9,10,11,12,13,14,18,19,22,32,33 only, and no transaction keys (24,
25, 26, 28) |
+| B3 |
Produce/Fetch/ListOffsets/CreateTopics/DeleteTopics/InitProducerId/OffsetCommit/OffsetFetch
out-of-range | N/A | **Connection closes** for both above-max and below-min -
`kafka_protocol`'s schema floor for each of these six messages equals
`SUPPORTED_RANGES`' own min, so there is no encodable error response below min
either (see `SCOPE.md`'s Governance model) |
+| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys
0,1,2,3,8,9,10,11,12,13,14,18,19,20,22,32,33 only, and no transaction keys (24,
25, 26, 28) |
An out-of-range version only ever produces `error_code=35` on ApiVersions
(B1); every other API
key's out-of-range case closes the connection - see B2/B3. InitProducerId and
Produce also send
@@ -256,7 +257,10 @@ decoded `ec`.
| C1 | 15 | DescribeGroups | `send --host 127.0.0.1:9093 --api-key 15` |
Connection closes, no response bytes |
| C2 | 10 | FindCoordinator | `send --host 127.0.0.1:9093 --api-key 10` |
Connection closes |
| C3 | 17 | SaslHandshake | `send --host 127.0.0.1:9093 --api-key 17` |
Connection closes |
-| C4 | 20 | DeleteTopics | `send --host 127.0.0.1:9093 --api-key 20` |
Connection closes |
+| C4 | 20 | DeleteTopics | `send --host 127.0.0.1:9093 --api-key 20` | Sends
v1-v6. Gateway answers v1-v5 (stub `NOT_CONTROLLER` (41) with the bridge off).
v6 is above the advertised max and that one connection closes |
+
+Key 20 is scoped, so C4 is not an unsupported-key close. `send --api-key 20`
walks v1-v6; the
+gateway answers v1-v5 and closes on v6, above the advertised max.
Reconnect and send A2 in a **new** session after each of C1-C4 to confirm the
*gateway* (not just
that one connection) is still serving other clients.
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 649f2ba71..0f1695466 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -2,14 +2,15 @@
## Issue #3421 — in scope (this iteration)
-A TCP listener on the Kafka wire port. It decodes requests, validates scoped
API keys, versions and wire formats, and answers them. With a bridge, Produce,
Fetch, ListOffsets, Metadata, CreateTopics, DescribeConfigs and AlterConfigs
use Iggy.
+A TCP listener on the Kafka wire port. It decodes requests, validates scoped
API keys, versions and wire formats, and answers them. With a bridge, Produce,
Fetch, ListOffsets, Metadata, CreateTopics, DeleteTopics, DescribeConfigs and
AlterConfigs use Iggy.
**Stub semantics (important):** without a bridge, every API answers with a
stub. Produce discards
the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). Fetch and
ListOffsets answer 6
-too. CreateTopics, DescribeConfigs and AlterConfigs answer `NOT_CONTROLLER`
(41), so clients do not believe
-topics were created or configs were read or changed. OffsetCommit and
OffsetFetch answer retriable `COORDINATOR_LOAD_IN_PROGRESS`
-(14). Do not read `ec=0` from a stub as durable storage. Produce is the one
API that stores records once you configure
-a bridge ([#3535](https://github.com/apache/iggy/issues/3535)).
+too. CreateTopics, DeleteTopics, DescribeConfigs and AlterConfigs answer
`NOT_CONTROLLER` (41), so
+clients do not believe a topic was created, deleted, or that configs were read
or changed.
+OffsetCommit and OffsetFetch answer retriable `COORDINATOR_LOAD_IN_PROGRESS`
+(14). Do not read `ec=0` from a stub as durable storage. Produce is the one
API that stores records
+once you configure a bridge
([#3535](https://github.com/apache/iggy/issues/3535)).
| Deliverable | Status | Location |
| ------------- | -------- | ---------- |
@@ -54,6 +55,7 @@ it knows the server supports flexible encoding.
| 1 | Fetch | 4 | 12 | 4, 5, 6, 7, 8, 9, 10, 11, 12 | With a bridge: shared
topic probes, paged `poll_messages` per partition with records, wait up to
`max_wait_ms` (max 18 s). Without one: stub returns `NOT_LEADER_OR_FOLLOWER`
(6) |
| 2 | ListOffsets | 1 | 6 | 1, 2, 3, 4, 5, 6 | Decode request; stub response |
| 19 | CreateTopics | 2 | 5 | 2, 3, 4, 5 | Decode request; stub returns
`NOT_CONTROLLER` (41); `-1` partitions/RF = broker default on v4+ |
+| 20 | DeleteTopics | 1 | 5 | 1, 2, 3, 4, 5 | With a bridge: deletes the Iggy
topic, never the backing stream (see `BRIDGE_MAPPING.md`). Without one: stub
returns `NOT_CONTROLLER` (41). v6 (topic-id-based) not advertised - this bridge
has no topic-id concept |
| 8 | OffsetCommit | 2 | 9 | 2 … 9 | With a bridge: stores each offset as an
Iggy external group offset, and a negative offset deletes it
([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). Without one:
`COORDINATOR_LOAD_IN_PROGRESS` (14). Flexible encoding at v8+ |
| 9 | OffsetFetch | 1 | 9 | 1 … 9 | With a bridge: reads each offset, -1 when
none is stored, and a null topic list reads every topic that holds the group.
Without one: 14. Flexible encoding at v6+ |
| 10 | FindCoordinator | 0 | 4 | 0, 1, 2, 3, 4 | Answers "this gateway" for
group keys; `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` (53) for the transaction
key type, `INVALID_REQUEST` (42) for share; flexible encoding at v3+ |
@@ -88,6 +90,7 @@ Use this table when configuring clients or generating wire
fixtures with `kafka-
| 14 | SyncGroup | 0–5 | v4 |
| 18 | ApiVersions | 0–3 | v3 |
| 19 | CreateTopics | 2–5 | v5 |
+| 20 | DeleteTopics | 1–5 | v4 |
| 22 | InitProducerId | 0–5 | v2 |
| 32 | DescribeConfigs | 1–4 | v4 |
| 33 | AlterConfigs | 0–2 | v2 |
@@ -105,7 +108,7 @@ All API keys not listed above close the connection (see
Governance model above)
| 29 | DescribeAcls | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised
only while it is on ([`ACL_MAPPING.md`](ACL_MAPPING.md)) |
| 36 | SaslAuthenticate | Implemented behind `IGGY_KAFKA_SASL_ENABLED`,
advertised only while it is on ([`AUTHENTICATION.md`](AUTHENTICATION.md)) |
| 24, 25, 26, 28 | AddPartitionsToTxn, AddOffsetsToTxn, EndTxn,
TxnOffsetCommit | Transactions - not supported, see below |
-| 20, 23, 27, 30, 31, 34, 35, 37+ | DeleteTopics, `OffsetForLeaderEpoch`,
`WriteTxnMarkers`, `CreateAcls`/`DeleteAcls`, IncrementalAlterConfigs (44),
etc. | Later issues. DescribeConfigs (32) and AlterConfigs (33) are supported |
+| 23, 27, 30, 31, 34, 35, 37+ | `OffsetForLeaderEpoch`, `WriteTxnMarkers`,
`CreateAcls`/`DeleteAcls`, IncrementalAlterConfigs (44), etc. | Later issues |
| 21 | DeleteRecords | Not advertised on purpose, see below |
| 68 | ConsumerGroupHeartbeat | KIP-848 protocol, opt-in via
`group.protocol=consumer`; the 4.0 default is still `classic` |
@@ -168,8 +171,8 @@ nor moved by retention. That is the same core change the
ListOffsets `EARLIEST`
| Layer | #3421 | Description |
| ------- | ------- | ------------- |
| **1 — Wire framing** | In scope | `server.rs` — custom, zero-copy frame I/O;
`header.rs` delegates version selection to `kafka_protocol::messages::ApiKey` |
-| **2 — Request/response codecs** | Partial | Decode/encode via the
`kafka_protocol` crate (broker feature only) for 14 keys; `bounds_guard.rs`
pre-validates against unbounded allocation before handing a frame to the crate;
stub responses except InitProducerId and the five consumer-group keys, and
Produce, Fetch, ListOffsets, Metadata, CreateTopics, DescribeConfigs and
AlterConfigs with a bridge |
-| **3 — Iggy bridge** | Produce, Fetch, ListOffsets, Metadata, CreateTopics,
DescribeConfigs and AlterConfigs wired | `bridge/` module (connection, topic
mapping, provisioning, high watermark, `topic_target` + `send_records`, `probe`
+ `poll`). Produce ([#3535](https://github.com/apache/iggy/issues/3535)), Fetch
([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets
([#3537](https://github.com/apache/iggy/issues/3537)), Metadata
([#3534](https://github.com/apache/iggy/issues/ [...]
+| **2 — Request/response codecs** | Partial | Decode/encode via the
`kafka_protocol` crate (broker feature only) for 15 keys; `bounds_guard.rs`
pre-validates against unbounded allocation before handing a frame to the crate;
stub responses except InitProducerId and the five consumer-group keys, and
Produce, Fetch, ListOffsets, Metadata, CreateTopics, DeleteTopics,
DescribeConfigs and AlterConfigs with a bridge |
+| **3 — Iggy bridge** | Produce, Fetch, ListOffsets, Metadata, CreateTopics,
DeleteTopics, DescribeConfigs and AlterConfigs wired | `bridge/` module
(connection, topic mapping, provisioning, high watermark, `topic_target` +
`send_records`, `probe` + `poll`). Produce
([#3535](https://github.com/apache/iggy/issues/3535)), Fetch
([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets
([#3537](https://github.com/apache/iggy/issues/3537)), Metadata
([#3534](https://github.com/apach [...]
---
@@ -238,6 +241,37 @@ below it are still open for the issues that build on top
of it.
request's own wire `timeout_ms` (clamped to `[1s, 30s]`) now bounds the
whole handler's
aggregate bridge work, not just decoded and discarded - a deadline that
fires answers every
topic `REQUEST_TIMED_OUT` rather than continuing to hold the shared
lockstep `IggyClient`.
+- [x] Real DeleteTopics ([#3546](https://github.com/apache/iggy/issues/3546)):
with
+ `IGGY_KAFKA_BRIDGE_ENABLED=true`, deletes the Iggy topic through
+ `IggyBridge::delete_kafka_topic`.
`src/protocol/handlers/delete_topics.rs`,
+ `tests/delete_topics_real_bridge_tests.rs`. With the bridge off, the
stub answers
+ `NOT_CONTROLLER` (41), matching CreateTopics' own stub convention.
+ - **Never deletes the backing Iggy stream**, even when the delete leaves it
with no topics in
+ it. `TopicMapping::resolve` sends every unmapped Kafka topic to the same
`default_stream`
+ (`ensure_stream_and_topic`'s own doc comment flags the identical sharing
risk for create), so
+ a stream this call happens to empty may still be the live home for a
different Kafka topic a
+ moment later - there is no way to tell the two cases apart from here, and
real Kafka itself
+ has no analogous container object for a `DeleteTopics` to clean up. See
+ [`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 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. 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):
- Iggy partitions are **0-based** (same as Kafka) — direct `partition_id`
mapping, no offset conversion
- Kafka consumer groups do **not** map onto Iggy consumer groups. Assignment
stays client-side, and Iggy's group registry is used as an offset key only
([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md))
diff --git a/gateways/kafka/scripts/ci-wire-fixtures.sh
b/gateways/kafka/scripts/ci-wire-fixtures.sh
index b6ca97836..5251e8392 100755
--- a/gateways/kafka/scripts/ci-wire-fixtures.sh
+++ b/gateways/kafka/scripts/ci-wire-fixtures.sh
@@ -24,7 +24,7 @@ set -euo pipefail
FIXTURES_DIR="gateways/kafka/tools/kafka-tool/kafka_messages"
# API keys requested by api_handler_tests, version_firewall_tests, and
server_e2e_tests.
-FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 19 22 32 33)
+FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 19 20 22 32 33)
usage() {
echo "Usage: $0 {generate|cleanup}" >&2
diff --git a/gateways/kafka/src/bridge/iggy_bridge/topics.rs
b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
index b7dda3335..b860af435 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/topics.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
@@ -260,6 +260,31 @@ impl IggyBridge {
}
}
+ /// Deletes the Iggy topic backing `kafka_topic`, resolved through the
configured
+ /// [`TopicMapping`](crate::bridge::topic_map::TopicMapping).
+ ///
+ /// Deliberately leaves the backing Iggy stream alone, even when this
delete empties it out:
+ /// `TopicMapping::resolve` sends every *unmapped* Kafka topic to the same
`default_stream`
+ /// (`ensure_stream_and_topic`'s own doc comment flags the same sharing
risk for create), so a
+ /// stream this call happens to empty may still be the live home for a
different Kafka topic a
+ /// moment later - this call has no way to tell the two cases apart, and
real Kafka itself has
+ /// no analogous container object for a `DeleteTopics` to clean up.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails
Kafka's own
+ /// topic-naming rules. Returns [`BridgeError::Timeout`] if a call takes
longer than
+ /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] with
`StreamIdNotFound` (1009) or
+ /// `TopicIdNotFound` (2010) if the mapped stream or topic doesn't exist,
or for
+ /// connectivity/auth failures.
+ pub async fn delete_kafka_topic(&self, kafka_topic: &str) -> Result<(),
BridgeError> {
+ validate_kafka_topic_name("kafka_topic", kafka_topic)?;
+ let (stream_name, topic_name) =
self.config.topic_mapping.resolve(kafka_topic);
+ let stream_id =
Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ let topic_id =
Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
+ with_request_timeout(self.client.delete_topic(&stream_id,
&topic_id)).await
+ }
+
/// Looks up `kafka_topic`, resolved through the configured
/// [`TopicMapping`](crate::bridge::topic_map::TopicMapping), without
creating it.
///
diff --git a/gateways/kafka/src/protocol/api.rs
b/gateways/kafka/src/protocol/api.rs
index b0f27feb5..c7fc01244 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -39,9 +39,9 @@ use crate::protocol::bounds_guard::{
};
use crate::protocol::handlers::init_producer_id::ProducerIdAllocator;
use crate::protocol::handlers::{
- alter_configs, api_versions, create_topics, decode_guarded,
describe_configs, dispatch, fetch,
- find_coordinator, heartbeat, init_producer_id, join_group, leave_group,
list_offsets, metadata,
- offset_commit, offset_fetch, produce, respond_or_close, sync_group,
+ alter_configs, api_versions, create_topics, decode_guarded, delete_topics,
describe_configs,
+ dispatch, fetch, find_coordinator, heartbeat, init_producer_id,
join_group, leave_group,
+ list_offsets, metadata, offset_commit, offset_fetch, produce,
respond_or_close, sync_group,
};
use crate::protocol::probe_board::ProbeBoard;
use crate::protocol::sasl::{
@@ -62,10 +62,11 @@ pub const API_KEY_SYNC_GROUP: i16 = 14;
pub const API_KEY_SASL_HANDSHAKE: i16 = 17;
pub const API_KEY_API_VERSIONS: i16 = 18;
pub const API_KEY_CREATE_TOPICS: i16 = 19;
-pub const API_KEY_DESCRIBE_CONFIGS: i16 = 32;
-pub const API_KEY_ALTER_CONFIGS: i16 = 33;
+pub const API_KEY_DELETE_TOPICS: i16 = 20;
pub const API_KEY_INIT_PRODUCER_ID: i16 = 22;
pub const API_KEY_DESCRIBE_ACLS: i16 = 29;
+pub const API_KEY_DESCRIBE_CONFIGS: i16 = 32;
+pub const API_KEY_ALTER_CONFIGS: i16 = 33;
pub const API_KEY_SASL_AUTHENTICATE: i16 = 36;
pub const DEFAULT_KAFKA_PORT: u16 = 9093;
@@ -285,6 +286,7 @@ static SUPPORTED_RANGES: &[ApiVersionRange] = &[
offset_fetch::RANGE,
api_versions::RANGE,
create_topics::RANGE,
+ delete_topics::RANGE,
describe_configs::RANGE,
alter_configs::RANGE,
init_producer_id::RANGE,
diff --git a/gateways/kafka/src/protocol/bounds_guard.rs
b/gateways/kafka/src/protocol/bounds_guard.rs
index 3d9f237e8..90c711fbf 100644
--- a/gateways/kafka/src/protocol/bounds_guard.rs
+++ b/gateways/kafka/src/protocol/bounds_guard.rs
@@ -703,6 +703,44 @@ pub fn validate_create_topics_shape(
Ok(())
}
+/// Mirrors the field order `DeleteTopicsRequest::decode` walks, for the
`topic_names`-based
+/// shape (v1-v5) only.
+///
+/// This bridge's `RANGE` never advertises v6, the `topics:
Vec<DeleteTopicState>`
+/// (topic-id-based) shape, so that branch is never reached and isn't walked
here.
+///
+/// # Errors
+///
+/// Returns an error when a declared array/string/bytes length cannot fit in
the bytes remaining
+/// in the frame, or the body is truncated or malformed in a way that cannot
be walked.
+pub fn validate_delete_topics_shape(
+ version: i16,
+ body: &Bytes,
+ max_frame_size: usize,
+) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), max_frame_size);
+ let flexible = version >= 4;
+
+ let topic_names_count = if flexible {
+ c.compact_array_count()?
+ } else {
+ c.legacy_array_count()?
+ };
+ for _ in 0..topic_names_count {
+ if flexible {
+ c.compact_string(false)?;
+ } else {
+ c.legacy_string(false)?;
+ }
+ }
+
+ let _timeout_ms = c.read_i32()?;
+ if flexible {
+ c.tagged_fields()?;
+ }
+ Ok(())
+}
+
/// Mirrors the field order `MetadataRequest::decode` walks.
///
/// # Errors
@@ -1580,6 +1618,41 @@ mod tests {
assert!(validate_create_topics_shape(5, &body,
TEST_MAX_FRAME_SIZE).is_err());
}
+ /// Same POC class as `create_topics_v5_huge_topics_count_rejected`,
against
+ /// `validate_delete_topics_shape`'s own `topic_names` compact array count.
+ #[test]
+ fn delete_topics_v5_huge_topic_names_count_rejected() {
+ let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF, 0x0F]);
+ assert!(validate_delete_topics_shape(5, &body,
TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ /// Same desync class as `create_topics_v5_desync_varint_poc_rejected`: a
5-byte over-long
+ /// varint consumed as the (empty) `topic_names` count, matching
`kafka_protocol`'s own 5-byte
+ /// cap. `DeleteTopics` has one fewer field than `CreateTopics` between
the array and the
+ /// request's tagged fields (`timeout_ms` only, no `validate_only`), so
the padding here is
+ /// four bytes, not five, to land the huge trailing varint on the
tagged-fields count instead.
+ #[test]
+ fn delete_topics_v5_desync_varint_poc_rejected() {
+ let body = Bytes::from_static(&[
+ 0x81, 0x80, 0x80, 0x80, 0x80, // 5-byte over-long varint, decodes
to count 0
+ 0x00, 0x00, 0x00, 0x00, // timeout_ms
+ 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // u32::MAX read as the
tagged-fields count
+ ]);
+ assert!(validate_delete_topics_shape(5, &body,
TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ /// Legacy (non-flexible) shape: a declared topic-name length with nothing
behind it. Covers
+ /// the `legacy_string` branch `validate_delete_topics_shape` walks below
v4, which the two
+ /// flexible-only POCs above never reach.
+ #[test]
+ fn delete_topics_v1_truncated_legacy_topic_name_rejected() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x00, 0x00, 0x01, // topic_names: legacy array count = 1
+ 0x03, 0xE8, // first name's declared length: 1000, no bytes follow
+ ]);
+ assert!(validate_delete_topics_shape(1, &body,
TEST_MAX_FRAME_SIZE).is_err());
+ }
+
/// Reviewed POC (C2): 1 topic x 65,535 partitions was within the *old*
65,536 cumulative
/// budget - no single array count exceeded `MAX_COLLECTION_LEN`, only
their product did -
/// and drove a 393 KB request to a 2.16 MB response built by one
synchronous,
diff --git a/gateways/kafka/src/protocol/handlers/create_topics.rs
b/gateways/kafka/src/protocol/handlers/create_topics.rs
index 760d39bc6..d43c09d10 100644
--- a/gateways/kafka/src/protocol/handlers/create_topics.rs
+++ b/gateways/kafka/src/protocol/handlers/create_topics.rs
@@ -18,7 +18,6 @@
//! `CreateTopics` (API key 19).
use std::collections::HashSet;
-use std::time::Duration;
use bytes::Bytes;
use iggy::prelude::IggyError;
@@ -42,8 +41,8 @@ use crate::protocol::handlers::topic_config::{
MAX_CONFIG_TOPICS, find_duplicate_names, topic_cap_message,
};
use crate::protocol::handlers::{
- decode_guarded, encode_message, handle_versioned_request,
is_supported_version,
- respond_or_close, unsupported_version_response,
+ clamp_request_timeout, decode_guarded, encode_message,
handle_versioned_request,
+ is_supported_version, respond_or_close, unsupported_version_response,
};
pub const RANGE: ApiVersionRange = ApiVersionRange {
@@ -66,21 +65,6 @@ const DEFAULT_PARTITION_COUNT: u32 = 1;
/// `NONE` for a partition count the real path can never actually create.
const MAX_PARTITIONS_COUNT: u32 = 1000;
-/// Bounds imposed on the request's own `timeout_ms` before it becomes the
aggregate bridge-work
-/// deadline. That value is client-supplied and otherwise unchecked: `0` or
negative would abort
-/// every topic on arrival, and an oversized one would tie up the shared
`IggyClient` past any
-/// reasonable request.
-const MIN_REQUEST_TIMEOUT: Duration = Duration::from_millis(1_000);
-const MAX_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
-
-/// Clamps the wire's own `timeout_ms` (KIP-4's field for exactly this) into
-/// `[MIN_REQUEST_TIMEOUT, MAX_REQUEST_TIMEOUT]` - unlike
`ListOffsets`/`Metadata`, `CreateTopics`
-/// carries a real client-supplied deadline to honor, not just a fixed
internal ceiling.
-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(
@@ -173,9 +157,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,
@@ -185,31 +173,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.as_str()) {
- 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.as_str()) {
+ 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);
@@ -750,19 +759,23 @@ mod tests {
}
#[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);
+ fn
find_duplicate_names_finds_a_name_repeated_across_two_requested_topics() {
+ let topics = [creatable_topic(1, 1), creatable_topic(1, 1)];
+ let names: Vec<&str> = topics.iter().map(|topic|
topic.name.as_str()).collect();
+ let duplicates = find_duplicate_names(names);
+ assert_eq!(duplicates, HashSet::from(["orders"]));
}
#[test]
- fn clamp_request_timeout_passes_through_a_reasonable_value_unchanged() {
- assert_eq!(clamp_request_timeout(5_000), Duration::from_secs(5));
+ fn find_duplicate_names_is_empty_when_every_name_is_unique() {
+ let topics = [
+ creatable_topic(1, 1),
+ CreatableTopic::default()
+ .with_name(topic_name("payments"))
+ .with_num_partitions(1)
+ .with_replication_factor(1),
+ ];
+ let names: Vec<&str> = topics.iter().map(|topic|
topic.name.as_str()).collect();
+ assert!(find_duplicate_names(names).is_empty());
}
}
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..bd41fd5ae
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/delete_topics.rs
@@ -0,0 +1,404 @@
+// 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.
+//!
+//! `DeletableTopicResult.error_message` is only encoded on the wire at v5+
(`kafka_protocol`'s
+//! own gate) - a client negotiating v1-v4 gets `error_code` only, with any
text this module sets
+//! 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;
+use kafka_protocol::messages::{DeleteTopicsRequest, DeleteTopicsResponse,
TopicName};
+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,
+ ERROR_TOPIC_AUTHORIZATION_FAILED, GatewayState, HandleOutcome,
+};
+use crate::protocol::bounds_guard::validate_delete_topics_shape;
+use crate::protocol::handlers::{
+ clamp_request_timeout, 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 topic names one `DeleteTopics` request may address through the
bridge.
+///
+/// Same rationale as `create_topics`'s own `MAX_CONFIG_TOPICS`-based cap, but
counted differently: every
+/// name here costs its own bridge round trip against the single lockstep
`IggyClient` every
+/// Kafka connection on this gateway shares, duplicates included, so this
charges the raw count
+/// rather than the distinct one `CreateTopics` charges.
+const MAX_BRIDGE_BACKED_TOPICS: usize = 100;
+
+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,
+ 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",
+ );
+ }
+ };
+
+ 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 {
+ tracing::warn!(
+ requested_topics = req.topic_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "DeleteTopics request addresses too many topics; rejecting"
+ );
+ let message = StrBytes::from(format!(
+ "this gateway addresses at most {MAX_BRIDGE_BACKED_TOPICS} topic
names 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");
+ }
+
+ // 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,
&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
+/// 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.
+///
+/// `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; \
+ 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 {
+ // The second arm: the write committed on its first attempt, the SDK's
own reconnect path
+ // replayed it, and the server's client-table dedup caught the replay
- not a fault. Same
+ // reasoning as `create_topics::create_one_topic`'s identical arm.
+ Ok(()) | Err(BridgeError::Iggy(IggyError::RequestAlreadyApplied)) => {
+ 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. A missing topic and an
unauthorized reply
+/// are routine, so they log at `debug!` with fixed text. Everything else gets
a fixed message
+/// and logs at `error!`.
+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(StrBytes::from(reason.clone())))
+ }
+ BridgeError::Iggy(
+ IggyError::StreamIdNotFound(_)
+ | IggyError::StreamNameNotFound(_)
+ | IggyError::TopicIdNotFound(_, _)
+ | IggyError::TopicNameNotFound(_, _)
+ | IggyError::PartitionNotFound(_, _, _)
+ | IggyError::ResourceNotFound(_),
+ ) => {
+ tracing::debug!(kafka_topic, error = %err, "DeleteTopics: topic
does not exist");
+ result
+ .with_error_code(error_code)
+ .with_error_message(Some(StrBytes::from(
+ "this topic does not exist".to_string(),
+ )))
+ }
+ BridgeError::Iggy(IggyError::Unauthorized) => {
+ tracing::debug!(
+ kafka_topic,
+ error = %err,
+ "DeleteTopics: not authorized to delete this topic"
+ );
+ result
+ .with_error_code(error_code)
+ .with_error_message(Some(StrBytes::from(
+ "not authorized to delete this topic".to_string(),
+ )))
+ }
+ other => {
+ tracing::error!(kafka_topic, %other, "DeleteTopics failed against
the Iggy bridge");
+ result
+ .with_error_code(error_code)
+ .with_error_message(Some(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 iggy::prelude::Identifier;
+
+ use super::*;
+
+ #[test]
+ fn bridge_error_result_names_a_missing_topic_and_an_unauthorized_reply() {
+ let missing_stream =
BridgeError::Iggy(IggyError::StreamIdNotFound(Identifier::default()));
+ let missing_topic = BridgeError::Iggy(IggyError::TopicIdNotFound(
+ Identifier::default(),
+ Identifier::default(),
+ ));
+ let unauthorized = BridgeError::Iggy(IggyError::Unauthorized);
+ let cases = [
+ (&missing_stream, "this topic does not exist"),
+ (&missing_topic, "this topic does not exist"),
+ (&unauthorized, "not authorized to delete this topic"),
+ ];
+ for (err, expected) in cases {
+ let result = bridge_error_result(DeletableTopicResult::default(),
"orders", err);
+ assert_eq!(result.error_code, err.to_kafka_error_code());
+ assert_eq!(result.error_message.as_deref(), Some(expected));
+ }
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs
b/gateways/kafka/src/protocol/handlers/mod.rs
index ea706f221..499b497f4 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -27,6 +27,7 @@
pub mod alter_configs;
pub mod api_versions;
pub mod create_topics;
+pub mod delete_topics;
pub mod describe_configs;
pub mod fetch;
pub mod find_coordinator;
@@ -42,6 +43,8 @@ pub mod produce;
pub mod sync_group;
pub(crate) mod topic_config;
+use std::time::Duration;
+
use bytes::{Buf, Bytes, BytesMut};
use iggy::prelude::IggyError;
use kafka_protocol::messages::TransactionalId;
@@ -53,12 +56,12 @@ use crate::auth::AuthenticatedPrincipal;
use crate::bridge::BridgeError;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS,
API_KEY_DESCRIBE_CONFIGS,
- API_KEY_FETCH, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT,
API_KEY_INIT_PRODUCER_ID,
- API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP, API_KEY_LIST_OFFSETS,
API_KEY_METADATA,
- API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH, API_KEY_PRODUCE,
API_KEY_SYNC_GROUP,
- ConnectionState, ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION,
GatewayState, HandleOutcome,
- is_supported_version, supported_max_version,
+ API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS,
API_KEY_DELETE_TOPICS,
+ API_KEY_DESCRIBE_CONFIGS, API_KEY_FETCH, API_KEY_FIND_COORDINATOR,
API_KEY_HEARTBEAT,
+ API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
API_KEY_LIST_OFFSETS,
+ API_KEY_METADATA, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH,
API_KEY_PRODUCE,
+ API_KEY_SYNC_GROUP, ConnectionState, ERROR_INVALID_REQUEST,
ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version, supported_max_version,
};
/// Record encodes and decodes of this many bytes or more run off the async
worker.
@@ -66,6 +69,20 @@ pub(crate) const CODEC_OFF_WORKER_BYTES: usize = 64 * 1024;
/// A plain copy costs less per byte, so the handoff pays off only from this
size.
pub(crate) const COPY_OFF_WORKER_BYTES: usize = 1024 * 1024;
+/// Client-supplied `timeout_ms` floor. `0` or negative would abort every
topic on arrival.
+pub(crate) const MIN_REQUEST_TIMEOUT: Duration = Duration::from_millis(1_000);
+/// Client-supplied `timeout_ms` ceiling. Above this the shared `IggyClient`
stays busy too long.
+pub(crate) const MAX_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Clamps KIP-4 `timeout_ms` into `[MIN_REQUEST_TIMEOUT,
MAX_REQUEST_TIMEOUT]`.
+///
+/// `CreateTopics` and `DeleteTopics` honor the request's own deadline.
`ListOffsets` and
+/// `Metadata` do not: they have no such field.
+pub(crate) 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)
+}
+
/// Runs `work`. When `heavy`, other tasks move to another worker first, so
they keep running. A
/// current-thread runtime runs it in place.
pub(crate) fn off_worker<T>(heavy: bool, work: impl FnOnce() -> T) -> T {
@@ -99,6 +116,7 @@ pub async fn dispatch(
API_KEY_OFFSET_FETCH => offset_fetch::handle(state, connection,
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, principal,
api_version, body).await,
API_KEY_DESCRIBE_CONFIGS => describe_configs::handle(state,
api_version, body).await,
API_KEY_ALTER_CONFIGS => alter_configs::handle(state, principal,
api_version, body).await,
API_KEY_FIND_COORDINATOR => find_coordinator::handle(state,
api_version, body).await,
@@ -289,4 +307,21 @@ mod tests {
async fn given_a_multi_thread_runtime_when_work_is_large_should_run_it() {
assert_eq!(off_worker(true, || 7), 7);
}
+
+ #[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/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/scope.rs
b/gateways/kafka/tests/common/scope.rs
index 4a9153713..63b4f38b9 100644
--- a/gateways/kafka/tests/common/scope.rs
+++ b/gateways/kafka/tests/common/scope.rs
@@ -39,6 +39,7 @@ pub const SCOPED_API_KEYS: &[(i16, &str, i16, i16)] = &[
(14, "SyncGroup", 0, 5),
(18, "ApiVersions", 0, 3),
(19, "CreateTopics", 2, 5),
+ (20, "DeleteTopics", 1, 5),
(22, "InitProducerId", 0, 5),
(32, "DescribeConfigs", 1, 4),
(33, "AlterConfigs", 0, 2),
diff --git a/gateways/kafka/tests/common/wire.rs
b/gateways/kafka/tests/common/wire.rs
index dc3b838a5..14c56db76 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -45,11 +45,7 @@ pub fn decode<M: Decodable>(body: Bytes, version: i16) -> M {
}
/// Consumer-group and admin keys explicitly out of scope in SCOPE.md.
-pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] = &[
- (15, "DescribeGroups"),
- (16, "ListGroups"),
- (20, "DeleteTopics"),
-];
+pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] = &[(15, "DescribeGroups"),
(16, "ListGroups")];
/// Append Metadata request fields that follow the topics array for `version`.
fn write_metadata_request_trailer(enc: &mut Encoder, version: i16) {
@@ -190,6 +186,47 @@ pub fn build_create_topics_empty_request(version: i16) ->
Bytes {
enc.freeze()
}
+/// Empty `DeleteTopics` request (no topic names) for supported versions
(v1-v5).
+pub fn build_delete_topics_empty_request(version: i16) -> Bytes {
+ let flexible = version >= 4;
+ let mut enc = Encoder::with_capacity(16);
+
+ if flexible {
+ enc.write_varint(1); // empty topic_names compact array (N+1 = 1)
+ } else {
+ enc.write_i32(0);
+ }
+ enc.write_i32(5_000); // timeout_ms
+ if flexible {
+ enc.write_empty_tagged_fields();
+ }
+
+ 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
new file mode 100644
index 000000000..f125d3f2a
--- /dev/null
+++ b/gateways/kafka/tests/delete_topics_real_bridge_tests.rs
@@ -0,0 +1,336 @@
+// 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_INVALID_REQUEST, 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, None, 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 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_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;
+ 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_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.
+#[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, 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);
+ assert!(
+ error_message.is_some_and(|m| m.contains("100")),
+ "error_message must name the actual cap"
+ );
+}
diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs
b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
index b020d6ea0..d1e763690 100644
--- a/gateways/kafka/tests/golden_wire_fixtures_tests.rs
+++ b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
@@ -41,11 +41,11 @@ async fn golden_apiversions_v3_flexible_response_fixture() {
.await
.expect_response("test request has acks != 0 and expects a response");
- // error_code=0, api_count=16 (compact array: N+1=17)
+ // error_code=0, api_count=17 (compact array: N+1=18)
// each entry followed by an empty tagged-fields byte; throttle_ms=0;
top-level tagged fields
- let expected: [u8; 120] = [
+ let expected: [u8; 127] = [
0x00, 0x00, // error_code
- 0x11, // compact array count (16+1)
+ 0x12, // compact array count (17+1)
0x00, 0x00, 0x00, 0x00, 0x00, 0x09, 0x00, // key 0: Produce
0-9 (advertised)
0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, 0x00, // key 1: Fetch
4-12
0x00, 0x02, 0x00, 0x01, 0x00, 0x06, 0x00, // key 2: ListOffsets
1-6
@@ -59,6 +59,7 @@ async fn golden_apiversions_v3_flexible_response_fixture() {
0x00, 0x0E, 0x00, 0x00, 0x00, 0x05, 0x00, // key 14: SyncGroup
0-5
0x00, 0x12, 0x00, 0x00, 0x00, 0x03, 0x00, // key 18: ApiVersions
0-3
0x00, 0x13, 0x00, 0x02, 0x00, 0x05, 0x00, // key 19: CreateTopics
2-5
+ 0x00, 0x14, 0x00, 0x01, 0x00, 0x05, 0x00, // key 20: DeleteTopics
1-5
0x00, 0x16, 0x00, 0x00, 0x00, 0x05, 0x00, // key 22: InitProducerId
0-5
0x00, 0x20, 0x00, 0x01, 0x00, 0x04, 0x00, // key 32: DescribeConfigs
1-4
0x00, 0x21, 0x00, 0x00, 0x00, 0x02, 0x00, // key 33: AlterConfigs
0-2
@@ -75,10 +76,10 @@ async fn golden_apiversions_v1_response_fixture() {
.await
.expect_response("test request has acks != 0 and expects a response");
- // error_code=0, api_count=16; Produce advertises min=0 per KAFKA-18659;
throttle_ms=0
- let expected: [u8; 106] = [
+ // error_code=0, api_count=17; Produce advertises min=0 per KAFKA-18659;
throttle_ms=0
+ let expected: [u8; 112] = [
0x00, 0x00, // error_code
- 0x00, 0x00, 0x00, 0x10, // api count = 16
+ 0x00, 0x00, 0x00, 0x11, // api count = 17
0x00, 0x00, 0x00, 0x00, 0x00, 0x09, // key 0: Produce 0-9
(advertised)
0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, // key 1: Fetch 4-12
0x00, 0x02, 0x00, 0x01, 0x00, 0x06, // key 2: ListOffsets 1-6
@@ -92,6 +93,7 @@ async fn golden_apiversions_v1_response_fixture() {
0x00, 0x0E, 0x00, 0x00, 0x00, 0x05, // key 14: SyncGroup 0-5
0x00, 0x12, 0x00, 0x00, 0x00, 0x03, // key 18: ApiVersions 0-3
0x00, 0x13, 0x00, 0x02, 0x00, 0x05, // key 19: CreateTopics 2-5
+ 0x00, 0x14, 0x00, 0x01, 0x00, 0x05, // key 20: DeleteTopics 1-5
0x00, 0x16, 0x00, 0x00, 0x00, 0x05, // key 22: InitProducerId 0-5
0x00, 0x20, 0x00, 0x01, 0x00, 0x04, // key 32: DescribeConfigs 1-4
0x00, 0x21, 0x00, 0x00, 0x00, 0x02, // key 33: AlterConfigs 0-2
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs
b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index 35cc5d713..ce835c239 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -1010,6 +1010,77 @@ fn list_acls(gateway: SocketAddr, username: &str,
password: &str) -> String {
.expect_ran(&format!("listing ACLs as {username}"))
}
+/// Runs `kafka-topics.sh` with `args` against `gateway`. No
`--command-config`: the bridged
+/// stack this test uses runs with SASL off, same as the kcat calls against it
above.
+fn kafka_topics(gateway: SocketAddr, args: &[&str]) -> ClientRun {
+ let bootstrap = gateway.to_string();
+ let mut full_args: Vec<&str> = vec![
+ "/opt/kafka/bin/kafka-topics.sh",
+ "--bootstrap-server",
+ &bootstrap,
+ ];
+ full_args.extend_from_slice(args);
+ run_client(KAFKA_IMAGE, &full_args, &[])
+}
+
+/// #3546's own acceptance criterion: `kafka-topics.sh` must work against the
gateway for create
+/// and delete. (`--describe` needs `DescribeConfigs`, not yet implemented, so
it is not covered
+/// here.) Runs against the bridged stack, since this exercises the real
`CreateTopics`/
+/// `DeleteTopics` bridge path, not authentication - the other Java-client
tests above already
+/// cover SASL.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn
given_kafka_topics_sh_when_creating_then_deleting_a_topic_should_round_trip_through_the_bridge()
+ {
+ const NEW_TOPIC: &str = "kafka-topics-roundtrip";
+ let Some((_server, gateway)) = bridged_stack().await else {
+ return;
+ };
+
+ let created = kafka_topics(
+ gateway,
+ &[
+ "--create",
+ "--topic",
+ NEW_TOPIC,
+ "--partitions",
+ "3",
+ "--replication-factor",
+ "1",
+ ],
+ )
+ .expect_ran("kafka-topics.sh --create");
+ assert!(
+ created.contains(&format!("Created topic {NEW_TOPIC}")),
+ "kafka-topics.sh must report the topic created, got: {created}"
+ );
+
+ let listed =
+ kafka_topics(gateway, &["--list"]).expect_ran("kafka-topics.sh --list
after create");
+ assert!(
+ listed.lines().any(|line| line.trim() == NEW_TOPIC),
+ "the created topic must appear in the listing, got: {listed}"
+ );
+
+ let deleted = kafka_topics(gateway, &["--delete", "--topic", NEW_TOPIC])
+ .expect_ran("kafka-topics.sh --delete");
+ assert!(
+ !deleted.to_lowercase().contains("error"),
+ "kafka-topics.sh --delete must succeed, got: {deleted}"
+ );
+
+ let listed_after =
+ kafka_topics(gateway, &["--list"]).expect_ran("kafka-topics.sh --list
after delete");
+ assert!(
+ !listed_after.lines().any(|line| line.trim() == NEW_TOPIC),
+ "the deleted topic must no longer appear in the listing, got:
{listed_after}"
+ );
+ assert!(
+ listed_after.lines().any(|line| line.trim() == TOPIC),
+ "deleting one topic must not take the shared default stream, and
`{TOPIC}`, with it, got: \
+ {listed_after}"
+ );
+}
+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn
given_records_produced_by_kcat_when_kcat_consumes_should_read_them_back_in_order()
{
let Some((_server, gateway)) = bridged_stack().await else {
diff --git a/gateways/kafka/tests/sasl_tests.rs
b/gateways/kafka/tests/sasl_tests.rs
index 2f6d5e3f8..476d0b84f 100644
--- a/gateways/kafka/tests/sasl_tests.rs
+++ b/gateways/kafka/tests/sasl_tests.rs
@@ -576,9 +576,9 @@ async fn
given_sasl_disabled_when_a_client_connects_should_serve_without_authent
let advertised = send(&mut stream, API_KEY_API_VERSIONS, 1, 2, &[]).await;
assert!(
- !advertised
- .windows(2)
- .any(|w| i16::from_be_bytes([w[0], w[1]]) ==
API_KEY_SASL_HANDSHAKE),
+ !parse_api_versions(&advertised)
+ .iter()
+ .any(|&(key, _, _)| key == API_KEY_SASL_HANDSHAKE),
"the SASL keys must not be advertised while the feature is off"
);
}
diff --git a/gateways/kafka/tests/version_firewall_tests.rs
b/gateways/kafka/tests/version_firewall_tests.rs
index 31240a70a..c3b7620f3 100644
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@ -37,12 +37,12 @@ use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use iggy_gateway_kafka::protocol::api::{
- API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS,
API_KEY_DESCRIBE_CONFIGS,
- API_KEY_FETCH, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT,
API_KEY_INIT_PRODUCER_ID,
- API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP, API_KEY_LIST_OFFSETS,
API_KEY_METADATA,
- API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH, API_KEY_PRODUCE,
API_KEY_SYNC_GROUP,
- ERROR_INVALID_REQUEST, ERROR_NONE, ERROR_UNSUPPORTED_VERSION,
advertised_min_version,
- handle_request, is_supported_version, supported_api_ranges,
+ API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS,
API_KEY_DELETE_TOPICS,
+ API_KEY_DESCRIBE_CONFIGS, API_KEY_FETCH, API_KEY_FIND_COORDINATOR,
API_KEY_HEARTBEAT,
+ API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
API_KEY_LIST_OFFSETS,
+ API_KEY_METADATA, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH,
API_KEY_PRODUCE,
+ API_KEY_SYNC_GROUP, ERROR_INVALID_REQUEST, ERROR_NONE,
ERROR_UNSUPPORTED_VERSION,
+ advertised_min_version, handle_request, is_supported_version,
supported_api_ranges,
};
use codec::Decoder;
@@ -57,17 +57,17 @@ use tcp::{
use wire::{
JoinGroupParams, OUT_OF_SCOPE_API_KEYS, SyncGroupParams,
build_alter_configs_empty_request,
build_api_versions_flexible_request, build_create_topics_empty_request,
- build_describe_configs_empty_request, build_fetch_empty_topics_request,
- build_find_coordinator_request, build_heartbeat_request,
build_init_producer_id_request,
- build_join_group_request, build_leave_group_request,
build_list_offsets_request,
- build_metadata_all_topics_flexible, build_metadata_all_topics_legacy,
- build_metadata_flexible_request_v10, build_offset_commit_request,
build_offset_fetch_request,
- build_sync_group_request,
+ build_delete_topics_empty_request, build_describe_configs_empty_request,
+ build_fetch_empty_topics_request, build_find_coordinator_request,
build_heartbeat_request,
+ build_init_producer_id_request, build_join_group_request,
build_leave_group_request,
+ build_list_offsets_request, build_metadata_all_topics_flexible,
+ build_metadata_all_topics_legacy, build_metadata_flexible_request_v10,
+ build_offset_commit_request, build_offset_fetch_request,
build_sync_group_request,
};
#[test]
-fn supported_ranges_table_has_sixteen_entries() {
- assert_eq!(supported_api_ranges().len(), 16);
+fn supported_ranges_table_has_seventeen_entries() {
+ assert_eq!(supported_api_ranges().len(), 17);
}
#[test]
@@ -104,7 +104,10 @@ fn is_supported_version_matches_scope_table() {
///
/// The wire rows are `supported_api_ranges()` sorted by `api_key`
(`api_versions.rs`).
/// `SCOPED_API_KEYS` is already in that ascending order, and this test reads
the response
-/// in that order. `supported_ranges_table_has_sixteen_entries` and
+/// in that order (Produce, Fetch, `ListOffsets`, Metadata, `OffsetCommit`,
`OffsetFetch`,
+/// `FindCoordinator`, `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`,
`ApiVersions`,
+/// `CreateTopics`, `DeleteTopics`, `InitProducerId`, `DescribeConfigs`,
`AlterConfigs`).
+/// `supported_ranges_table_has_seventeen_entries` and
/// `is_supported_version_matches_scope_table` pin that both tables cover the
same keys.
#[tokio::test]
async fn apiversions_advertises_exact_supported_ranges_v1() {
@@ -496,6 +499,7 @@ fn request_body_for_scoped_api(api_key: i16, name: &str,
version: i16) -> Bytes
.flatten()
.unwrap_or_else(|| build_list_offsets_request(version,
"scope-topic", 0)),
API_KEY_CREATE_TOPICS => build_create_topics_empty_request(version),
+ API_KEY_DELETE_TOPICS => build_delete_topics_empty_request(version),
API_KEY_FIND_COORDINATOR => build_find_coordinator_request(version,
&["scope-group"], 0),
// A member id the coordinator never handed out, so every in-range
version answers
// UNKNOWN_MEMBER_ID immediately instead of creating a group and
parking on its join
